Now I want to show how it is possible to achieve a similar flow control using specialized listeners that allow to manipulate the
ServerSession message queue.
ServerSession message queue is a data structure that is accessed concurrently when messages are published and delivered to clients, so it needs appropriate synchronization when accessed.
In order to simplify this synchronization requirements, CometD allows you to add
ServerSessions, with the guarantee that these listeners will be called with the appropriate locks acquired, to allow user code to freely modify the queue’s content.
Below you can find an example of a
DeQueueListener that keeps only the first message of a series of message published to the same channel within a tolerance period of 1000 ms, and removes the others (it relies on the timestamp extension):
String channelName = "/stock/GOOG";
long tolerance = 1000;
ServerSession session = ...;
public void deQueue(ServerSession session, Queue<serverMessage> queue)
long lastTimeStamp = 0;
for (Iterator<serverMessage> iterator = queue.iterator(); iterator.hasNext();)
ServerMessage message = iterator.next();
long timeStamp = Long.parseLong(message.get(Message.TIMESTAMP_FIELD).toString());
if (timeStamp <= lastTimeStamp + tolerance)
System.err.println("removed " + message);
System.err.println("kept " + message);
lastTimeStamp = timeStamp;
Other possibilities include keeping the last message (instead of the first), coalescing the message fields following a particular logic, or even clearing the queue completely.
DeQueueListeners are called when CometD is about to deliver messages to the client, so clearing the queue completely results in an empty response being sent to the client.
This is different from the behavior of lazy channels, that allowed to delay the message delivery until a configurable timeout expired.
However, lazy channels do not alter the number of messages being sent, while
DeQueueListeners can manipulate the message queue.
Therefore, CometD message control flow is often best accomplished by using both mechanisms: lazy channels to delay message delivery, and
DeQueueListeners to reduce/coalesce the number of messages sent.