Class CqcSubscriptionHandler

java.lang.Object

com.loomcache.server.handler.CqcSubscriptionHandler

public final class CqcSubscriptionHandler
extends Object

Server-side handler for the CQC (Continuous Query Cache) wire protocol.

Clients send MessageType.CQC_SUBSCRIBE with a SerializablePredicate encoded in the message value; this handler registers a DistributedMap.MapChangeListener on the target map that evaluates the predicate server-side. Matched events are pushed back to the client as MessageType.CQC_EVENT messages; non-matching events are silently dropped, saving bandwidth compared to the legacy MessageType.LISTENER_EVENT fan-out-then-client-filter path.

Lifecycle

• CQC_SUBSCRIBE: register/replace the listener for (peerId, mapName) and respond OK.
• CQC_UNSUBSCRIBE: deregister the listener and respond OK.
• Client disconnect: onPeerDisconnect(String) tears down all subscriptions for the peer.

Event wire format

CQC_EVENT is laid out like LISTENER_EVENT (for parity) with the map name in mapName, the key in key, and the value being:

  [2-byte eventType len][eventType UTF-8][4-byte oldValue len][oldValue UTF-8][4-byte newValue len][newValue UTF-8]

Null values are encoded with length -1 and no payload bytes.

Thread safety

Subscription state uses concurrent collections; listener dispatch happens on whatever thread mutated the map (same contract as DistributedEventListenerManager).

Since:

2.0

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	CqcSubscriptionHandler.CqcSubscription	Active CQC subscription.
static final record	CqcSubscriptionHandler.SubscriptionId	Composite key identifying a single subscription.

Constructor Summary

Constructors

Constructor	Description
CqcSubscriptionHandler(DataStructureRegistry registry, TcpServer tcpServer, int instanceNumber)

Method Summary

Modifier and Type	Method	Description
int	getActiveSubscriptionCount()	Number of active subscriptions (observability only).
Map<CqcSubscriptionHandler.SubscriptionId, CqcSubscriptionHandler.CqcSubscription>	getSubscriptions()	Snapshot of active subscriptions, keyed by (peerId, mapName).
Message	handle(String peerId, Message msg)	Dispatch a CQC_SUBSCRIBE or CQC_UNSUBSCRIBE message from the given peer.
static boolean	isCqcOperation(@Nullable MessageType type)	Returns true for message types this handler owns.
void	onPeerDisconnect(String peerId)	Remove all subscriptions for a disconnected peer.

Methods inherited from class Object

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

Constructor Details

CqcSubscriptionHandler

public CqcSubscriptionHandler(DataStructureRegistry registry,
 TcpServer tcpServer,
 int instanceNumber)

Method Details

isCqcOperation

public static boolean isCqcOperation(@Nullable MessageType type)

Returns true for message types this handler owns.

handle

public Message handle(String peerId,
 Message msg)

Dispatch a CQC_SUBSCRIBE or CQC_UNSUBSCRIBE message from the given peer.

Parameters:

peerId - the source peer id (for delivery of CQC_EVENT pushes)

msg - the CQC message

Returns:

the response to send back to the peer

onPeerDisconnect

public void onPeerDisconnect(String peerId)

Remove all subscriptions for a disconnected peer.

getActiveSubscriptionCount

public int getActiveSubscriptionCount()

Number of active subscriptions (observability only).

getSubscriptions

public Map<CqcSubscriptionHandler.SubscriptionId, CqcSubscriptionHandler.CqcSubscription> getSubscriptions()

Snapshot of active subscriptions, keyed by (peerId, mapName). Exposed for tests.