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 Summary

    Constructors
    Constructor
    Description
    InMemoryMessageStore(io.fluxzero.common.MessageType messageType)
     

  • Method Summary

    Modifier and Type
    Method
    Description
    default CompletableFuture<Void>
    append(io.fluxzero.common.api.SerializedMessage... arg0)
     
    CompletableFuture<Void>
    append(List<io.fluxzero.common.api.SerializedMessage> messages)
     
    protected void
    appendMessages(List<io.fluxzero.common.api.SerializedMessage> messages)
    Stores messages while the caller owns this store's monitor.
    void
    close()
     
    protected Collection<io.fluxzero.common.api.SerializedMessage>
    filterMessages(Collection<io.fluxzero.common.api.SerializedMessage> messages)
     
    default List<io.fluxzero.common.api.SerializedMessage>
    getBatch(Long arg0, int arg1)
     
    List<io.fluxzero.common.api.SerializedMessage>
    getBatch(Long minIndex, int maxSize, boolean inclusive)
     
    List<io.fluxzero.common.api.SerializedMessage>
    getBatch(Long minIndex, int maxSize, boolean inclusive, long maxBytes)
     
    default List<io.fluxzero.common.api.SerializedMessage>
    getBatch(Long arg0, int arg1, long arg2)
     
    protected io.fluxzero.common.api.SerializedMessage
    getMessage(long index)
     
    default io.fluxzero.common.tracking.MessageStore
    getMessageStore()
     
    protected Collection<io.fluxzero.common.api.SerializedMessage>
    messagesFrom(Long minIndex, boolean inclusive)
     
    protected final Object
    monitorNotificationLock()
    Serializes a store mutation with its subsequent monitor notification while allowing callbacks to re-enter the store itself.
    void
    notifyMonitors()
    Wakes registered monitors with an empty message batch, without appending messages.
    protected void
    notifyMonitors(List<io.fluxzero.common.api.SerializedMessage> messages)
     
    protected void
    purgeExpiredMessages(Duration messageExpiration)
     
    io.fluxzero.common.Registration
    registerMonitor(Consumer<List<io.fluxzero.common.api.SerializedMessage>> monitor)
     
    io.fluxzero.common.tracking.MessageStoreBatch
    scanBatch(Long minIndex, int maxSize, boolean inclusive, long maxBytes, Predicate<? super io.fluxzero.common.api.SerializedMessage> filter)
     
    default io.fluxzero.common.tracking.MessageStoreBatch
    scanBatch(Long arg0, int arg1, boolean arg2, long arg3, Predicate<? super io.fluxzero.common.api.SerializedMessage> arg4, boolean arg5)
     
    void
    setRetentionTime(Duration arg0)
     
    String
    toString()
     
    void
    truncate()
     
    protected void
    truncateMessages()
    Clears the message log while the caller owns this store's monitor.
    default <T extends io.fluxzero.common.tracking.MessageStore>
    T
    unwrap(Class<T> arg0)
     

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

  • Constructor Details

    • InMemoryMessageStore

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

  • Method Details

    • append

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

    • appendMessages

      protected void appendMessages(List<io.fluxzero.common.api.SerializedMessage> messages)
      Stores messages while the caller owns this store's monitor. Subclasses can use this to update their secondary indexes and the message log atomically without invoking external monitors under the store lock.

    • getBatch

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

    • getBatch

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

    • scanBatch

      public io.fluxzero.common.tracking.MessageStoreBatch scanBatch(Long minIndex, int maxSize, boolean inclusive, long maxBytes, Predicate<? super io.fluxzero.common.api.SerializedMessage> filter)

    • notifyMonitors

      public void notifyMonitors()
      Wakes registered monitors with an empty message batch, without appending messages.

    • 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)

    • messagesFrom

      protected Collection<io.fluxzero.common.api.SerializedMessage> messagesFrom(Long minIndex, boolean inclusive)

    • getMessage

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

    • truncate

      public void truncate()

    • truncateMessages

      protected void truncateMessages()
      Clears the message log while the caller owns this store's monitor.

    • monitorNotificationLock

      protected final Object monitorNotificationLock()
      Serializes a store mutation with its subsequent monitor notification while allowing callbacks to re-enter the store itself. Subclasses must acquire this lock before the store monitor whenever both are needed.

    • 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)

    • getBatch

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

    • scanBatch

      default io.fluxzero.common.tracking.MessageStoreBatch scanBatch(Long arg0, int arg1, boolean arg2, long arg3, Predicate<? super io.fluxzero.common.api.SerializedMessage> arg4, boolean arg5)

    • 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()