Named Objects and Shared Named Objects

Universal Messaging provides the ability for the server to maintain state for the last event that was consumed by a consumer on a channel. By providing a unique name, you can create a so-called named object on a channel. Then, even if your application is stopped and restarted, you will only consume available events from the last event ID that the server stored as successfully consumed by that named object.

When using normal named objects (as opposed to shared named objects, see below for description), there is only one store of events and that is the channel itself. Each event remains on the channel until all named objects have consumed (and acknowledged) that event, and then the event is removed from the channel.

Named objects can be persistent, i.e. the last event ID is written to disk, so that if the Universal Messaging Realm Server is restarted, the last event ID consumed is retrievable for each named object on a channel.

A UM channel (in JMS terminology this would be a "topic") can have several different "engines", one of which is the JMS engine. The JMS engine means that the channel will only store events whilst they are required by a consumer. So if you have a slow synchronous consumer, the channel will store the events until they are pulled off. If you have a named object, the channel will store the events until they are acknowledged. If you use the normal fanout engine, the events will remain on the channel regardless of any named objects. Only capacity or TTL or purges will delete events.

One reason for using named objects is to facilitate recovery if a consumer is terminated. For example, if the consumer crashes or you turn off the computer and start up another one, the server will remember where that consumer finished off because it stored the event ID on the named object.

Normally, UM maintains a store of events for each named object (known in JMS as a "durable subscription") until a consumer has received the events. When the consumer informs UM that it has received a certain event from the store, UM deletes this event from the store and all other events in the store that have a lower event ID (i.e. were created at an earlier time). There can be no more than 1 consumer of a given named object at any time; if a consumer tries to access a named object that is already being accessed by another consumer, an error will be returned to the second consumer.

There are some cases in which it may be desirable for two or more consumers to access a named object at the same time. UM offers a variation of a named object known as a shared named object.

A shared named object allows multiple consumers to access the same named object. UM delivers events to the consumers in round-robin fashion. So, for example, if there are 10 events in the named object and two consumers who share the named object, then UM will deliver events 1, 3, 5, 7, 9 to the first consumer and events 2, 4, 6, 8, 10 to the second consumer.

As for normal named objects, you can specify that the shared named object is "persistent" and/or "cluster wide". If the shared named object is persistent, UM will retain it if there is a system restart. If the shared named object is cluster wide and it exists on a cluster wide channel, the named object will also be replicated across the cluster.

The Universal Messaging client API offers the option of specifying a window size for a shared named object. The window size can be set independently for both synchronous and asynchronous subscribers which use shared named objects. 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 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 named object is not a shared named object, 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 asynchronous subscribers this is "ClientQueueWindow", which is available in the "Cluster Config" group; for synchronous subscribers, this is "IteratorWindowSize", which is available in the "Fanout Values" configuration group.

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

nSessionAttributes sessionAttr = new nSessionAttributes("nsp://localhost:11000");
nSession session = nSessionFactory.create(sessionAttr);
session.init();

nChannelAttributes channelAttr = new nChannelAttributes("channel");
channel = session.createChannel(channelAttr);
boolean isPersistent = true;
boolean isClusterWide = false;
int startEventId = 0;
nNamedObject namedObject = channel.createSharedNamedObject("shared-named-object",
isPersistent, isClusterWide, startEventId);
nEventListener listener = new nEventListener() {
@Override
public void go(nConsumeEvent evt) {
System.out.println("Consumed event " + evt.getEventID());
}
};
int windowSize = 4;
channel.addSubscriber(listener, namedObject, null, windowSize);

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

nSessionAttributes sessionAttr = new nSessionAttributes("nsp://localhost:11000");
nSession session = nSessionFactory.create(sessionAttr);
session.init();

nChannelAttributes channelAttr = new nChannelAttributes("channel");
channel = session.createChannel(channelAttr);
boolean isPersistent = true;
boolean isClusterWide = false;
int startEventId = 0;
nNamedObject namedObject = channel.createSharedNamedObject("shared-named-object",
isPersistent, isClusterWide, startEventId);
int windowSize = 4;
nChannelIterator iterator = channel.createIterator(namedObject, null, windowSize);

nChannelIterator supports an option for specifying an infinite window size, namely channel.createIterator(namedObject, null, nChannelIterator.INFINITE_WINDOW).

When you define a shared named object, UM maintains a copy of all of the events of the named object 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 named object, all earlier events are also removed; however, for a shared named object, only the consumed event is removed from the store. This is by design, and can, in time, lead to UM retaining large numbers of events that are never used. If you have made the shared named object 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 named object is created on. That purge will also flow through to the internal store.

You can also use the setting of the UM global parameter SharedDurableFilterBound to limit the set of events that are copied to the internal store for a given shared named object. If you set the value of this parameter to "true", the following behavior will be activated:

When the first consumer connects to the shared named object and specifies an event filter, then only events that match the filter will be copied to the internal store.

If a second consumer connects to the same shared named object 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 still uses its filter to take events off the internal store.

You can use the API method getSharedNamedObjectOutstandingEvents() on nNamedObject to get the number of events that are outstanding for the shared durable, i.e. the number of messages that have not yet been acknowledged by the shared durable and are therefore residing on the server. If you call this method on a non-shared durable, it will return the value -1.