Class CachingTrackingClient

java.lang.Object
io.fluxzero.sdk.tracking.client.CachingTrackingClient
All Implemented Interfaces:
TrackingClient, AutoCloseable
public class CachingTrackingClient extends Object implements TrackingClient
A TrackingClient implementation that wraps another client (typically a WebsocketTrackingClient) and caches recent messages in memory to reduce redundant round trips to the Fluxzero Runtime.

This client is particularly useful in environments where multiple consumers or trackers are processing the same stream of messages. Rather than each tracker reading from the backend individually, a shared in-memory cache serves recent messages directly when possible.

Behavior

  • Internally starts a special tracker that continuously appends new messages to a bounded in-memory cache.
  • Trackers that read from this client are first served from the local cache when possible.
  • Falls back to the delegate TrackingClient for uncached or missed messages.
  • Falls back to the delegate when the cache has no immediately available messages.
  • Cache size is limited via maxCacheSize; old messages are evicted in insertion order.

Use Cases

  • Optimizing performance when many trackers are polling the same stream concurrently
  • Reducing network latency and load on the Fluxzero Runtime for high-volume message types
  • Minimizing end-to-end processing delay in horizontally scaled applications

Tracking Mechanics

  • Uses a background tracker configured with ignoreSegment = true and clientControlledIndex = true to stream all new messages into the cache.
  • Trackers calling read(String, Long, ConsumerConfiguration) are served from the cache if their lastIndex is already present and the next batch is available in memory.
  • If not, the delegate is queried directly to maintain completeness and runtime-side lifecycle behavior.

Thread Safety

  • The cache is backed by a ConcurrentSkipListMap for safe concurrent access.
  • Eviction is synchronized to prevent race conditions.
See Also:

  • Constructor Details

    • CachingTrackingClient

      public CachingTrackingClient(WebsocketTrackingClient delegate)

    • CachingTrackingClient

      public CachingTrackingClient(TrackingClient delegate, int maxCacheSize)

  • Method Details

    • read

      public CompletableFuture<MessageBatch> read(String trackerId, Long lastIndex, ConsumerConfiguration config)
      Description copied from interface: TrackingClient
      Asynchronously reads the next available MessageBatch for a given tracker.
      Specified by:
      read in interface TrackingClient
      Parameters:
      trackerId - the unique ID for the tracker thread requesting messages
      lastIndex - the last index successfully handled by this tracker
      config - the full configuration for the consumer
      Returns:
      a CompletableFuture that completes with the next batch of messages

    • waitForCachedBatch

      protected CompletableFuture<MessageBatch> waitForCachedBatch(ConsumerConfiguration config, long minIndex, ClaimSegmentResult claim, Instant deadline)

    • doWaitForCachedBatch

      protected MessageBatch doWaitForCachedBatch(ConsumerConfiguration config, long minIndex, ClaimSegmentResult claim, Instant deadline) throws InterruptedException
      Throws:
      InterruptedException

    • getMessageBatch

      protected MessageBatch getMessageBatch(ConsumerConfiguration config, long minIndex, ClaimSegmentResult claim)

    • filterMessages

      protected List<SerializedMessage> filterMessages(List<SerializedMessage> messages, int[] segmentRange, Position position, ConsumerConfiguration config)

    • filterPredicate

      protected Predicate<SerializedMessage> filterPredicate(int[] segmentRange, Position position, ConsumerConfiguration config)

    • cacheNewMessages

      protected void cacheNewMessages(List<SerializedMessage> messages)

    • removeOldMessages

      protected void removeOldMessages()

    • readFromIndex

      public List<SerializedMessage> readFromIndex(long minIndex, int maxSize)
      Description copied from interface: TrackingClient
      Fetches messages starting from the given index up to the provided max size.

      This method bypasses consumer configurations and is primarily used for diagnostics or reprocessing.

      Specified by:
      readFromIndex in interface TrackingClient
      Parameters:
      minIndex - the starting index (inclusive)
      maxSize - the maximum number of messages to retrieve
      Returns:
      a list of serialized messages starting at the given index

    • readFromIndex

      public List<SerializedMessage> readFromIndex(long minIndex, int maxSize, long maxBytes)
      Description copied from interface: TrackingClient
      Fetches messages directly from the message log while limiting both message count and serialized payload bytes.

      If the first available message is larger than maxBytes, it is still returned so consumers can make progress. A maxBytes value of 0 disables the byte limit.

      Specified by:
      readFromIndex in interface TrackingClient
      Parameters:
      minIndex - the starting index (inclusive)
      maxSize - the maximum number of messages to retrieve
      maxBytes - the maximum number of serialized payload bytes to retrieve
      Returns:
      a list of serialized messages starting at the given index

    • readRange

      public List<SerializedMessage> readRange(long minIndexInclusive, long maxIndexExclusive, int maxSize)
      Description copied from interface: TrackingClient
      Fetches messages in the range [minIndexInclusive, maxIndexExclusive) up to the provided max size.

      Implementations may override this to perform a bounded server-side range read. The default implementation keeps the wire protocol backward compatible by using TrackingClient.readFromIndex(long, int) and filtering locally.

      Specified by:
      readRange in interface TrackingClient
      Parameters:
      minIndexInclusive - the starting index (inclusive)
      maxIndexExclusive - the maximum index (exclusive)
      maxSize - the maximum number of messages to retrieve
      Returns:
      a list of serialized messages in the requested range

    • readRange

      public List<SerializedMessage> readRange(long minIndexInclusive, long maxIndexExclusive, int maxSize, long maxBytes)
      Description copied from interface: TrackingClient
      Fetches messages in the range [minIndexInclusive, maxIndexExclusive) while limiting both message count and serialized payload bytes.
      Specified by:
      readRange in interface TrackingClient
      Parameters:
      minIndexInclusive - the starting index (inclusive)
      maxIndexExclusive - the maximum index (exclusive)
      maxSize - the maximum number of messages to retrieve
      maxBytes - the maximum number of serialized payload bytes to retrieve
      Returns:
      a list of serialized messages in the requested range

    • claimSegment

      public CompletableFuture<ClaimSegmentResult> claimSegment(String trackerId, Long lastIndex, ConsumerConfiguration config)
      Description copied from interface: TrackingClient
      Claims a processing segment for the given tracker.

      Segments are used to partition the message log among multiple tracker threads for parallel processing.

      Specified by:
      claimSegment in interface TrackingClient
      Parameters:
      trackerId - the unique identifier of the tracker attempting to claim a segment
      lastIndex - the tracker's last successfully processed index
      config - the full consumer configuration
      Returns:
      a CompletableFuture resolving to the result of the claim

    • storePosition

      public CompletableFuture<Void> storePosition(String consumer, int[] segment, long lastIndex, Guarantee guarantee)
      Description copied from interface: TrackingClient
      Stores the last successfully processed position for a consumer with a specific delivery guarantee.
      Specified by:
      storePosition in interface TrackingClient
      Parameters:
      consumer - the name of the consumer
      segment - the segment the tracker is processing
      lastIndex - the last message index processed
      guarantee - delivery guarantee (e.g., STORED, SENT)
      Returns:
      a future indicating completion

    • resetPosition

      public CompletableFuture<Void> resetPosition(String consumer, long lastIndex, Guarantee guarantee)
      Description copied from interface: TrackingClient
      Resets the consumer's tracking position to a given index with a specific delivery guarantee.
      Specified by:
      resetPosition in interface TrackingClient
      Parameters:
      consumer - the name of the consumer
      lastIndex - the new index to start from
      guarantee - the delivery guarantee
      Returns:
      a future indicating completion

    • getPosition

      public Position getPosition(String consumer)
      Description copied from interface: TrackingClient
      Returns the current committed tracking position for the given consumer.
      Specified by:
      getPosition in interface TrackingClient
      Parameters:
      consumer - the name of the consumer
      Returns:
      the last known committed position

    • disconnectTracker

      public CompletableFuture<Void> disconnectTracker(String consumer, String trackerId, boolean sendFinalEmptyBatch, Guarantee guarantee)
      Description copied from interface: TrackingClient
      Disconnects the specified tracker from its segment with the specified delivery guarantee.
      Specified by:
      disconnectTracker in interface TrackingClient
      Parameters:
      consumer - the name of the consumer group
      trackerId - the ID of the tracker thread being disconnected
      sendFinalEmptyBatch - whether to send a final empty batch to commit state
      guarantee - the delivery guarantee to use
      Returns:
      a future indicating disconnection

    • getMessageType

      public MessageType getMessageType()
      Description copied from interface: TrackingClient
      Returns the MessageType (e.g., COMMAND, EVENT, QUERY) associated with this tracking client.
      Specified by:
      getMessageType in interface TrackingClient
      Returns:
      the message type

    • getTopic

      public String getTopic()
      Description copied from interface: TrackingClient
      Returns the topic associated with this tracking client.

      This is applicable only when TrackingClient.getMessageType() is MessageType.DOCUMENT or MessageType.CUSTOM, where messages are organized into named topics beyond the standard type-based categorization.

      For other MessageTypes (e.g., COMMAND, EVENT), the concept of a topic is implicit and not required for tracking.

      Specified by:
      getTopic in interface TrackingClient
      Returns:
      the topic name, or null if not applicable for the message type

    • close

      public void close()
      Description copied from interface: TrackingClient
      Closes any open resources associated with this client.

      Once closed, the client should no longer be used to fetch or commit tracking state.

      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface TrackingClient