Class InMemoryMessageStore

java.lang.Object
io.fluxzero.sdk.tracking.client.InMemoryMessageStore
All Implemented Interfaces:
AutoCloseable
Direct Known Subclasses:
InMemoryEventStore, InMemoryScheduleStore
public class InMemoryMessageStore extends Object
An in-memory implementation of the MessageStore interface for storing SerializedMessages without external persistence.

This store underpins both local tracking (via LocalTrackingClient) and local publishing (via in-memory GatewayClient) in test and development environments.

Behavior

  • Messages are assigned a unique, incrementing index upon append if none is present.
  • Stored messages are retained in memory using a ConcurrentSkipListMap keyed by index.
  • Supports expiration via retentionTime, with periodic purging during appends.
  • Supports message monitors that are notified after every append.

Thread Safety

  • Append and monitor notifications are synchronized to preserve consistency across batch inserts.
  • Message storage is based on concurrent data structures, safe for multi-threaded access.
  • Monitors use a CopyOnWriteArraySet for thread-safe iteration and updates.

Use Cases

  • Unit and integration tests for consumers, handlers, and gateways
  • Simulating message flow in local environments without a Fluxzero backend
  • Standalone tools that mock message streams

Message Expiration

  • Expired messages are purged based on wall-clock time via Fluxzero.currentTime().
  • The purge logic is triggered during each call to append(List) when a retention policy is set.
See Also:

  • Constructor Details

    • InMemoryMessageStore

      public InMemoryMessageStore(io.fluxzero.common.MessageType messageType)

  • Method Details

    • append

      public CompletableFuture<Void> append(List<io.fluxzero.common.api.SerializedMessage> messages)

    • getBatch

      public List<io.fluxzero.common.api.SerializedMessage> getBatch(Long minIndex, int maxSize, boolean inclusive)

    • notifyMonitors

      public void notifyMonitors()

    • notifyMonitors

      protected void notifyMonitors(List<io.fluxzero.common.api.SerializedMessage> messages)

    • purgeExpiredMessages

      protected void purgeExpiredMessages(Duration messageExpiration)

    • filterMessages

      protected Collection<io.fluxzero.common.api.SerializedMessage> filterMessages(Collection<io.fluxzero.common.api.SerializedMessage> messages)

    • getMessage

      protected io.fluxzero.common.api.SerializedMessage getMessage(long index)

    • registerMonitor

      public io.fluxzero.common.Registration registerMonitor(Consumer<List<io.fluxzero.common.api.SerializedMessage>> monitor)

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • append

      default CompletableFuture<Void> append(io.fluxzero.common.api.SerializedMessage... arg0)

    • getBatch

      default List<io.fluxzero.common.api.SerializedMessage> getBatch(Long arg0, int arg1)

    • setRetentionTime

      void setRetentionTime(Duration arg0)

    • unwrap

      default <T extends io.fluxzero.common.tracking.MessageStore> T unwrap(Class<T> arg0)

    • getMessageStore

      default io.fluxzero.common.tracking.MessageStore getMessageStore()