Using Durable Subscriptions with Multiple Clients

The Universal Messaging client API offers the option of specifying a window size for a shared durable subscription. The window size can be set independently for both synchronous and asynchronous subscribers which use shared durable subscriptions. The client API contains two methods that accept the window size as a parameter: one for creating an iterator for a channel and one for adding a new subscriber with nEventListener.

When specifying the window size, the session is by default not auto-acknowledge. This means that when the window size is reached, the client stops receiving more events if the subscription is asynchronous. When an iterator is created and the window size is exceeded, the client is asked to commit or roll back.

If the API parameter for window size is not specified, nConstants.setMaxUnackedEvents(x) will be used as the default client window size for all asynchronous subscribers, if this value has been set before subscribing with listeners. If the durable subscription is not a shared durable subscription, an nIllegalArgumentException is thrown.

If you specify a negative window size for this API parameter, the default value stored in the server configurations will be used instead. For Shared-Queued synchronous durable subscribers, this is "IteratorWindowSize", which is available in the "Fanout Values" configuration group; for all other cases of durable subscription, the default value is "ClientQueueWindow", which is available in the "Cluster Config" group.

You can set the size of the client window using a method from the public nChannel API: channel.addSubscriber(nEventListener nel, nDurable name, String selector, int windowSize). A different window size can be configured for every asynchronous subscriber. Here is a simple code snippet for this:

nChannelAttributes channelAttr = new nChannelAttributes("channel");
nChannel channel = session.createChannel(channelAttr);
boolean isPersistent = true;
boolean isClusterWide = false;
int startEventId = 0;

nDurableAttributes attr =
nDurableAttributes.create(nDurableAttributes.nDurableType.Shared,
"shared-durable");

attr.setPersistent(isPersistent);
attr.setClustered(isClusterWide);
attr.setStartEID(startEventId);

nDurable sharedDurable = null;

// Need to check first in case shared durable already exists!
try {
sharedDurable = channels.getDurableManager().get(attr.getName());
} catch (nNameDoesNotExistException ex) {
sharedDurable = channels.getDurableManager().add(attr);
}

nEventListener listener = new nEventListener() {
@Override
public void go(nConsumeEvent evt) {
System.out.println("Consumed event " + evt.getEventID());
}
};

int windowSize = 4;
channel.addSubscriber(listener, sharedDurable, null, windowSize);

You can also set the size of the client window when creating nChannelIterator from a channel object: channel.createIterator(nDurable name, String selector, int windowSize). Here is an example:

nSession session = nSessionFactory.create(sessionAttr);
session.init();

nChannelAttributes channelAttr = new nChannelAttributes("channel");
nChannel channel = session.createChannel(channelAttr);
boolean isPersistent = true;
boolean isClusterWide = false;
int startEventId = 0;
nDurable sharedDurable = null;

// create the shared durable
nDurableAttributes attr =
nDurableAttributes.create(nDurableAttributes.nDurableType.Shared,
"shared-durable");

attr.setPersistent(isPersistent);
attr.setClustered(isClusterWide);
attr.setStartEID(startEventId);

// Need to check first in case shared durable is already exist!
try {
sharedDurable = channels.getDurableManager().get(attr.getName());
} catch (nNameDoesNotExistException ex) {
sharedDurable = channels.getDurableManager().add(attr);
}

int windowSize = 4;
nChannelIterator iterator =
channel.createIterator(sharedDurable, null, windowSize);

nRealmNode realmNode =
new nRealmNode(new nSessionAttributes("nsp://localhost:9000"));
nTopicNode topicNode = (nTopicNode) realmNode.findNode(CHANNEL_NAME);

nDurableNode durableNode = topicNode.getDurable(DURABLE_NAME);
long eventCount = durableNode.getDepth();

long unackedCount = durableNode.getTransactionDepth();

When you define a shared durable subscription, Universal Messaging maintains a copy of all of the events of the durable subscription in an internal store. The copies remain in the internal store until the original events have been accessed by the consumer. The benefit for the consumer is that events are held until the consumer is ready to receive them.

If the consumer accesses the events using event filtering, some events are never selected, and therefore the copies of these events remain in the internal store. As mentioned earlier, when an event is consumed for a normal durable subscription, all earlier events are also removed; however, for a shared durable subscription, only the consumed event is removed from the store. This is by design, and can, in time, lead to Universal Messaging retaining large numbers of events that are never used. If you have made the shared durable subscription persistent, a copy of the internal store will be maintained on disk, and this disk copy will also contain copies of the events that are not used.

In order to delete messages from the internal store you can purge events from the channel that the durable subscription is created on. That purge will also flow through to the internal store.

If a second consumer connects to the same shared durable subscription and also specifies an event filter, then from that time onwards, only events matching the second filter will be copied to the internal store. The first consumer receives an asynchronous exception due to the changed filter, but still uses its original filter to take events off the internal store.