Package io.fluxcapacitor.common.tracking

Class DefaultTrackingStrategy

java.lang.Object

io.fluxcapacitor.common.tracking.DefaultTrackingStrategy

All Implemented Interfaces:

TrackingStrategy, Closeable, AutoCloseable

public class DefaultTrackingStrategy
extends Object
implements TrackingStrategy

Streaming strategy that allows multiple clients to concurrently consume a message stream. Messages are routed to clients based on the value of their segment. Each connected client handles a distinct range of segments.

Message segments are determined by the clients that publish the messages (usually based on the consistent hash of some routing key, like the value of a user id).

If a client joins or leaves the cluster the segment range mapped to each client is recalculated so messages may get routed differently than before.

Clients can safely join or leave the cluster at any time. The strategy guarantees that a message is not consumed by more than one client.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected class	DefaultTrackingStrategy.WaitingTracker

Constructor Summary

Constructors

Modifier	Constructor	Description
	DefaultTrackingStrategy(MessageStore source)
	DefaultTrackingStrategy(MessageStore source, TaskScheduler scheduler)
protected	DefaultTrackingStrategy(MessageStore source, TaskScheduler scheduler, int segments)

Method Summary

Modifier and Type	Method	Description
protected int	adjustMaxSize(Tracker tracker, int maxSize)
protected int[]	claimSegment(Tracker tracker)
void	claimSegment(Tracker tracker, PositionStore positionStore)
void	close()
void	disconnectTrackers(Predicate<Tracker> predicate, boolean sendFinalEmptyBatch)
protected SerializedMessage	ensureMessageSegment(SerializedMessage message)
protected List<SerializedMessage>	filter(List<SerializedMessage> messages, int[] segmentRange, Position position, Tracker tracker)
protected List<SerializedMessage>	getBatch(int[] segment, Position position, int batchSize)
void	getBatch(Tracker tracker, PositionStore positionStore)
protected void	onClusterUpdate(TrackerCluster cluster)
protected void	onUpdate(List<SerializedMessage> messages)
protected Position	position(Tracker tracker, PositionStore positionStore, int[] segment)
protected void	purgeCeasedTrackers(Duration delay)
protected void	waitForMessages(Tracker tracker, MessageBatch emptyBatch, PositionStore positionStore)
protected void	waitForUpdate(Tracker tracker, MessageBatch emptyBatch, Runnable followUp)

Methods inherited from class java.lang.Object

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

Constructor Details

DefaultTrackingStrategy

public DefaultTrackingStrategy(MessageStore source)

DefaultTrackingStrategy

public DefaultTrackingStrategy(MessageStore source,
 TaskScheduler scheduler)

DefaultTrackingStrategy

protected DefaultTrackingStrategy(MessageStore source,
 TaskScheduler scheduler,
 int segments)

Method Details

getBatch

public void getBatch(Tracker tracker,
 PositionStore positionStore)

Specified by:

getBatch in interface TrackingStrategy

claimSegment

public void claimSegment(Tracker tracker,
 PositionStore positionStore)

Specified by:

claimSegment in interface TrackingStrategy

getBatch

protected List<SerializedMessage> getBatch(int[] segment,
 Position position,
 int batchSize)

waitForMessages

protected void waitForMessages(Tracker tracker,
 MessageBatch emptyBatch,
 PositionStore positionStore)

waitForUpdate

protected void waitForUpdate(Tracker tracker,
 MessageBatch emptyBatch,
 Runnable followUp)

position

protected Position position(Tracker tracker,
 PositionStore positionStore,
 int[] segment)

filter

protected List<SerializedMessage> filter(List<SerializedMessage> messages,
 int[] segmentRange,
 Position position,
 Tracker tracker)

ensureMessageSegment

protected SerializedMessage ensureMessageSegment(SerializedMessage message)

adjustMaxSize

protected int adjustMaxSize(Tracker tracker,
 int maxSize)

claimSegment

protected int[] claimSegment(Tracker tracker)

onUpdate

protected void onUpdate(List<SerializedMessage> messages)

onClusterUpdate

protected void onClusterUpdate(TrackerCluster cluster)

disconnectTrackers

public void disconnectTrackers(Predicate<Tracker> predicate,
 boolean sendFinalEmptyBatch)

Specified by:

disconnectTrackers in interface TrackingStrategy

purgeCeasedTrackers

protected void purgeCeasedTrackers(Duration delay)

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Specified by:

close in interface TrackingStrategy