Module io.fluxzero.sdk

Package io.fluxzero.sdk.tracking.client

Class InMemoryMessageStore

java.lang.Object

io.fluxzero.sdk.tracking.client.InMemoryMessageStore

All Implemented Interfaces:

io.fluxzero.common.Monitored<List<io.fluxzero.common.api.SerializedMessage>>, io.fluxzero.common.tracking.HasMessageStore, io.fluxzero.common.tracking.MessageStore, AutoCloseable

Direct Known Subclasses:

InMemoryEventStore, InMemoryScheduleStore

public class InMemoryMessageStore
extends Object
implements io.fluxzero.common.tracking.MessageStore

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:

• MessageStore
• LocalTrackingClient

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method	Description
CompletableFuture<Void>	append(List<io.fluxzero.common.api.SerializedMessage> messages)
void	close()
protected Collection<io.fluxzero.common.api.SerializedMessage>	filterMessages(Collection<io.fluxzero.common.api.SerializedMessage> messages)
List<io.fluxzero.common.api.SerializedMessage>	getBatch(Long minIndex, int maxSize, boolean inclusive)
protected io.fluxzero.common.api.SerializedMessage	getMessage(long index)
void	notifyMonitors()
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)
String	toString()

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.fluxzero.common.tracking.MessageStore

append, getBatch, getMessageStore, setRetentionTime, unwrap

Constructor Details

InMemoryMessageStore

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

Method Details

append

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

Specified by:

append in interface io.fluxzero.common.tracking.MessageStore

getBatch

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

Specified by:

getBatch in interface io.fluxzero.common.tracking.MessageStore

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)

Specified by:

registerMonitor in interface io.fluxzero.common.Monitored<List<io.fluxzero.common.api.SerializedMessage>>

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface io.fluxzero.common.tracking.MessageStore

toString

public String toString()

Overrides:

toString in class Object