Class ReplicationStream

java.lang.Object

com.loomcache.server.replication.ReplicationStream

public class ReplicationStream
extends Object

Continuous replication stream from leader to a single read replica.

Manages sending committed log entries in order, with backpressure handling and automatic reconnection on disconnect with exponential backoff.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	ReplicationStream.SendEntryCallback	Callback interface for sending entries to the replica.
static final record	ReplicationStream.StreamStats	Aggregated stats about the replication stream.

Constructor Summary

Constructors

Constructor	Description
ReplicationStream(String replicaId, String replicaAddress, int instanceNumber)	Initialize a replication stream.

Method Summary

Modifier and Type	Method	Description
void	advanceBackoff()	Advance the backoff (exponential backoff on disconnect).
long	getCurrentBackoffMs()	Get the current backoff time in milliseconds.
long	getNextIndexToSend()	Get the next index to send.
ReplicationStream.StreamStats	getStats()	Get aggregated stream statistics.
boolean	isConnected()	Check if the stream is connected.
boolean	isRunning()	Check if the stream is running.
void	resetBackoff()	Reset the backoff (on successful connection).
boolean	sendEntry(long index, long term, byte[] data)	Send an entry to the replica, handling backpressure.
void	setSendCallback(ReplicationStream.SendEntryCallback callback)	Set the callback for sending entries.
void	start()	Start the replication stream.
void	stop()	Stop the replication stream.
void	updateNextIndex(long newIndex)	Update the next index to send (e.g., after ACK from replica).

Methods inherited from class Object

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

Constructor Details

ReplicationStream

public ReplicationStream(String replicaId,
 String replicaAddress,
 int instanceNumber)

Initialize a replication stream.

Parameters:

replicaId - The replica identifier

replicaAddress - The replica address

instanceNumber - The instance number for logging

Method Details

setSendCallback

public void setSendCallback(ReplicationStream.SendEntryCallback callback)

Set the callback for sending entries.

Parameters:

callback - The send callback

start

public void start()

Start the replication stream.

stop

public void stop()

Stop the replication stream.

sendEntry

public boolean sendEntry(long index,
 long term,
 byte[] data)

Send an entry to the replica, handling backpressure.

Parameters:

index - The log entry index

term - The log entry term

data - The serialized entry data

Returns:

true if queued or sent successfully, false if backpressure or not running

getNextIndexToSend

public long getNextIndexToSend()

Get the next index to send.

Returns:

The next log index

updateNextIndex

public void updateNextIndex(long newIndex)

Update the next index to send (e.g., after ACK from replica).

Parameters:

newIndex - The new next index

isConnected

public boolean isConnected()

Check if the stream is connected.

Returns:

true if connected

isRunning

public boolean isRunning()

Check if the stream is running.

Returns:

true if running

advanceBackoff

public void advanceBackoff()

Advance the backoff (exponential backoff on disconnect).

resetBackoff

public void resetBackoff()

Reset the backoff (on successful connection).

getCurrentBackoffMs

public long getCurrentBackoffMs()

Get the current backoff time in milliseconds.

Returns:

current backoff delay

getStats

public ReplicationStream.StreamStats getStats()

Get aggregated stream statistics.

Returns:

StreamStats record