Named Objects, Shared Named Objects and Serial 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 and serial 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.

A Universal Messaging 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.

Each read operation issued by a client returns a single event from the durable to the client. The client processes the event according to the client's application logic, then returns an acknowledgement to the durable that the event has been processed. Each subsequent read operation by the client returns the next event from the durable in chronological order.

The following diagram shows the behavior of the Exclusive durable. Client session C1 is subscribed and connected to the durable. No other client sessions can subscribe to the same durable while the current client session is active and subscribed.

The following diagram illustrates the behavior of the Shared and Serial durable types, which can have multiple active client sessions. The client sessions are served in a round-robin manner. Each client receives a window of events. The Shared type differs from the Serial type in how the clients process their assigned event windows.

With the Shared durable type, multiple clients can subscribe and receive events from the durable in round-robin fashion. You can configure the durable so that each client receives a window that consists of multiple events instead of a single event. As soon as one client has received its allotted event or window of events, the next client receives its events, and so on. All subscribed clients on the durable can process the events concurrently.

The chronological order in which events are processed is not guaranteed. For example: the C1 client receives a window of events, then the C2 client receives a window of events, but C2 completes its processing and acknowledges its events before C1 completes its processing.

Multiple client sessions can subscribe and receive events from the durable, but the delivery method ensured that only one client will receive events at a time. The Serial type differs from the Shared type in that a client will only receive an event or a window of events when the previous client has acknowledged or rolled back its events. While any of the subscribed clients is processing its events, the other clients are inactive.

With a single client, Software AG recommends setting the window size to one to guarantee the serial delivery and processing of events. The server will not send more events to the client until the event that is already sent has been acknowledged or rolled back. Note that when "Window Size=1", the event delivery and processing is not done in parallel and the delivery throughput is lower.

The window size is set to 1. The client acknowledges events 1 to 4, but rolls back event 5. The server will wait for each event (1,2,3, and 4) to be processed before the server re-sends event 5. This ensures that events 1 to 5 are processed in the exact order, in which they were sent and received.

When the size of the event window is larger than 1, the server performance improves, because the delivery and processing of the batches of events is done in parallel. However, the serial delivery of events is not guaranteed in the following cases:

When an event is rolled back, the client receives the next event from the window (which was sent during the rollback), instead of the event that was rolled back. In this case the order in which the events get delivered to the client does not match the sending order.

The window size is set to 6. The client acknowledges events 1 to 4, but rolls back event 5. If the server has already sent event 6, that event will be delivered next instead of the rolled-back event. Event 5 will get re-delivered after event 6.

The events are delivered to a client after they have been processed by the previous client. Only one client can have unprocessed events at a time. If none of the clients have outstanding events, the delivery can be scheduled for any of the available clients. The events are delivered among clients in a round-robin fashion.

When the server attempts a round-robin delivery with an event window larger than 1, the round-robin delivery is not guaranteed. For example, if a client processes the events in parallel with the event delivery from the server, the client might receive more than one consequent batches of events.

C1 receives all six events and acknowledges event 1. When a new event is published to the server, the server delivers the event to C1, because C1 has an empty seat for 1 event in its window and 5 unprocessed events. To ensure the serial processing of the events, the server will send events to C2 only after C1 processes all of its in-flight events.

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.

For serial named objects, there is a special case when there is only one client session subscribed: if this client acknowledges some, but not all, of the events in the current window, Universal Messaging immediately sends the client the same number of new events from the next window instead of waiting until the client has acknowledged all of the events in the current window. This means that the client again has a number of unacknowledged events equal to the size of a window. Example: assume the window size is 10 and there is only one client session. If the client acknowledges 4 events from the first window of 10 events, Universal Messaging immediately sends 4 new events from the second window, so that the client is again working with a full set of 10 unacknowledged events. This pattern is repeated every time the client acknowledges events.

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");
nChannel channel = session.createChannel(channelAttr);
boolean isPersistent = true;
boolean isClusterWide = false;
int startEventId = 0;

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

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

nDurable namedObject = null;

// Need to check first in case shared durable already exists!
try {
namedObject = channels.getDurableManager().get(attr.getName());
} catch (nNameDoesNotExistException ex) {
namedObject = 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, namedObject, null, windowSize);

nSessionAttributes sessionAttr = new nSessionAttributes("nsp://localhost:11000");
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 namedObject = null;

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

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

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

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

When you define a shared named object, Universal Messaging 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 Universal Messaging 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.

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.