Maven requires you to set up the JAVA_HOME environment variable to point to your JDK installation.

After that, running the CometD demos is very simple. Assuming that $COMETD is the CometD installation directory, and that you have the mvn executable in your path:

$ cd $COMETD
$ cd cometd-demo
$ mvn jetty:deploy-war

The last command starts an embedded Jetty that listens on port 8080. Now point your browser to http://localhost:8080, to see the CometD Demos main page.

In the $COMETD/cometd-demo/target directory you can find a deployable WAR file that you can deploy in any Servlet (version 2.5 or greater) Container. Refer to the Servlet Container configuration manual to learn how to deploy the WAR in your Servlet Container. Refer also to Section 8.2.1.5, “Configuring Servlet 3 Asynchronous Features” for further information and for specific instructions related to deployment on Servlet 3.0 (or greater) Containers.

To enable debug logging on the JavaScript client library (see Chapter 7, JavaScript Library) you must pass the logLevel field to the configuration of the cometd object, with value 'debug' (see also Section 7.1, “Configuring and Initializing” for other configuration options):

cometd.configure({
    url: 'http://localhost:8080/cometd',
    logLevel: 'debug'
});

The CometD JavaScript library uses the window.console object to output log statements.

Once you have logging enabled on the JavaScript client library, you can see the debug logs in the browser JavaScript console. For example, in Firefox 24 you can open the JavaScript console by clicking on Tools → Web Developer → Web Console, while in Chrome 30 you can open the JavaScript console by clicking on Tools → Developer Tools.

To enable debug logging on the Java server library (see Section 8.2, “Server Library”), you must pass the logLevel option to the configuration, with the value 3 (see also Section 8.2.1, “Configuring the Java Server” for further information):

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <init-param>
            <param-name>logLevel</param-name>
            <param-value>3</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

</web-app>

The CometD Java libraries (both client and server) use SLF4J as logging framework.

By default, SLF4J comes with with a simple binding that does not log statements produced at debug level. Therefore you must configure SLF4J with a more advanced binding such as Log4J or Logback. Refer to the SLF4J documentation and the binding implementation documentation on how to configure logging when using the advanced bindings.

As a simple example, for Log4J it is usually enough to add the SLF4J-Log4J binding jar (slf4j-log4j12-<slf4j-version>.jar) and the Log4J jar (log4j-<log4j-version>.jar) in WEB-INF/lib, and then add a properly configured log4j.properties file in WEB-INF/classes. You don't want to add these files manually; they are reported here just as a simple reference. Usually these dependencies are configured in the build system that you use to create your web application.

Once you have logging enabled on the Java server library, you can see the debug logs in the terminal where you launched your Servlet Container (or in its log files, depending on how it is configured).

Setting up a project based on the CometD libraries using Maven uses the Maven archetypes, which create the skeleton of the project, in a style very similar to Rails scaffolding.

Issue the following command from a directory that does not contain a pom.xml file (otherwise you will get a Maven error), for example an empty directory:

$ cd /tmp
$ mvn archetype:generate -DarchetypeCatalog=http://cometd.org
...
Choose archetype:
1: http://cometd.org -> cometd-archetype-dojo-jetty6
2: http://cometd.org -> cometd-archetype-jquery-jetty6
3: http://cometd.org -> cometd-archetype-dojo-jetty7
4: http://cometd.org -> cometd-archetype-jquery-jetty7
5: http://cometd.org -> cometd-archetype-spring-jquery-jetty7
6: http://cometd.org -> cometd-archetype-spring-dojo-jetty7
Choose a number:

As you can see, there are six archetypes available that build a skeleton application using the Dojo or jQuery JavaScript toolkits, both with the choice of using Jetty 6 or Jetty 7 and Spring. Choose Dojo with Jetty 7, which is archetype number 3. The archetype generation requires that you define several properties and generates the application skeleton for you, for example:

Choose a number: : 3
Define value for property 'groupId': : org.cometd.primers
Define value for property 'artifactId': : dojo-jetty7-primer
Define value for property 'version':  1.0-SNAPSHOT: :
Define value for property 'package':  org.cometd.primers: :
[INFO] Using property: cometdVersion = 2.3.1
[INFO] Using property: jettyVersion = 7.4.4.v20110707
Confirm properties configuration:
groupId: org.cometd.primers
artifactId: dojo-jetty7-primer
version: 1.0-SNAPSHOT
package: org.cometd.primers
cometdVersion: 2.3.1
jettyVersion: 7.4.4.v20110707
Y: :
...
[INFO] BUILD SUCCESS

Then:

$ cd dojo-jetty7-primer/

The skeleton project now exists as follows:

$ tree .
.
|-- pom.xml
`-- src
`-- main
|-- java
|   `-- org
|       `-- cometd
|           `-- primers
|               |-- BayeuxInitializer.java
|               `-- HelloService.java
`-- webapp
|-- WEB-INF
|   `-- web.xml
|-- application.js
`-- index.jsp

The skeleton project is ready for you to run using the following command:

$ mvn install jetty:run
...

Now point your browser at http://localhost:8080/dojo-jetty7-primer, and you should see this message:

CometD Connection Succeeded
Server Says: Hello, World

That's it. You have already written your first CometD application :-)

The first step is to configure your favorite JavaScript toolkit, in our example Dojo, that the web container must serve. Using the Maven Way, this is obtained automatically by overlaying the CometD Dojo bindings WAR file, cometd-javascript-dojo-[version].war, but here you must do it manually. The cometd-javascript-dojo-[version].war is located in the cometd-javascript/dojo/target directory of the CometD distribution.

The final content, equivalent to that produced by the Maven way, should be like this:

.
|-- dijit
|-- dojo
|-- dojox
|   |-- cometd
|   |   |-- ack.js
|   |   |-- reload.js
|   |   |-- timestamp.js
|   |   `-- timesync.js
|   `-- cometd.js
|-- org
|   |-- cometd
|   |   |-- AckExtension.js
|   |   |-- ReloadExtension.js
|   |   |-- TimeStampExtension.js
|   |   `-- TimeSyncExtension.js
|   `-- cometd.js
|-- WEB-INF
|   |-- classes
|   |   `-- org
|   |       `-- cometd
|   |           `-- primers
|   |               |-- BayeuxInitializer.class
|   |               `-- HelloService.class
|   |-- lib
|   |   |-- bayeux-api-[version].jar
|   |   |-- cometd-java-common-[version].jar
|   |   |-- cometd-java-server-[version].jar
|   |   |-- jetty-continuation-[version].jar
|   |   |-- jetty-jmx-[version].jar
|   |   |-- jetty-servlets-[version].jar
|   |   `-- jetty-util-[version].jar
|   `-- web.xml
|-- application.js
`-- index.jsp

The org directory contains the new shared CometD implementation and the shared extensions, while the correspondent files in the dojox directory are the Dojo bindings. Other bindings exist for the jQuery toolkit, but the shared CometD implementation is the same.

The second step is to configure the server side. If you use Java, this means that you have to set up the CometD servlet that responds to messages from clients. The details of the server side configuration and service development are explained in Section 8.2, “Server Library”.

The last step is to write a JSP (or HTML) file that downloads the JavaScript dependencies and the JavaScript application, as explained in the following section.

The JSP file, index.jsp, contains the reference to the JavaScript toolkit dependencies and to the JavaScript application file:

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd">
<html>
<head>
<script type="text/javascript" src="${pageContext.request.contextPath}/dojo/dojo.js.uncompressed.js"></script>
<script type="text/javascript" src="application.js"></script>
<script type="text/javascript">
var config = {
contextPath: '${pageContext.request.contextPath}'
};
</script>
</head>
<body>
...
</body>
</html>

It also configures a JavaScript configuration object, config, with variables that the JavaScript application might need. This is totally optional.

The JavaScript application, contained in the application.js file, configures the cometd object and starts the application. The archetypes provide:

require(['dojo/dom', 'dojo/_base/unload', 'dojox/cometd', 'dojo/domReady!'],
function(dom, unloader, cometd)
{
    function _connectionEstablished()
    {
        dom.byId('body').innerHTML += '<div>CometD Connection Established</div>';
    }

    function _connectionBroken()
    {
        dom.byId('body').innerHTML += '<div>CometD Connection Broken</div>';
    }

    function _connectionClosed()
    {
        dom.byId('body').innerHTML += '<div>CometD Connection Closed</div>';
    }

    // Function that manages the connection status with the Bayeux server
    var _connected = false;
    function _metaConnect(message)
    {
        if (cometd.isDisconnected())
        {
            _connected = false;
            _connectionClosed();
            return;
        }

        var wasConnected = _connected;
        _connected = message.successful === true;
        if (!wasConnected && _connected)
        {
            _connectionEstablished();
        }
        else if (wasConnected && !_connected)
        {
            _connectionBroken();
        }
    }

    // Function invoked when first contacting the server and
    // when the server has lost the state of this client
    function _metaHandshake(handshake)
    {
        if (handshake.successful === true)
        {
            cometd.batch(function()
            {
                cometd.subscribe('/hello', function(message)
                {
                    dom.byId('body').innerHTML += '<div>Server Says: ' + message.data.greeting + '</div>';
                });
                // Publish on a service channel since the message is for the server only
                cometd.publish('/service/hello', { name: 'World' });
            });
        }
    }

    // Disconnect when the page unloads
    unloader.addOnUnload(function()
    {
        cometd.disconnect(true);
    });

    var cometURL = location.protocol + "//" + location.host + config.contextPath + "/cometd";
    cometd.configure({
        url: cometURL,
        logLevel: 'debug'
    });

    cometd.addListener('/meta/handshake', _metaHandshake);
    cometd.addListener('/meta/connect', _metaConnect);

    cometd.handshake();
});

Notice the following:

A channel is a string that looks like a URL path such as /foo/bar, /meta/connect or /service/chat.

The Bayeux specification defines three types of channels: meta channels, service channels and broadcast channels (see Appendix C, The Bayeux Protocol Specification v1.0).

A channel that starts with /meta/ is a meta channel, a channel that starts with /service/ is a service channel, and all other channels are broadcast channels.

A message whose channel field is a meta channel is referred to as a meta message, and similarly there are service messages and broadcast messages.

The application creates service channels and broadcast channels; an application can create as many as it needs, and can do so at any time.

Sessions are a central concept in CometD. They are the representation of the half-objects involved in the protocol communication.

There are three types of sessions:

The server is represented by an instance of org.cometd.bayeux.server.BayeuxServer. The BayeuxServer object acts as a:

Applications use listeners to interact with sessions, channels and the server. The Java and JavaScript APIs allow applications to register different kinds of listeners that receive notifications of the correspondent events. You can usefully think of extensions, security policies and authorizers as special types of listeners. The following sections treat them as such.

This section describes message processing on both the client and the server. Use the following image to understand the detailed components view that make up the client and the server.

When a client sends messages, it uses the client-side channel to publish them. The client retrieves the client channel from the client session via ClientSession.getChannel(String). Messages first pass through the extensions, which process messages one by one; if one extension denies processing of a message, it is deleted and it is not sent to the server. At the end of extension processing, the messages pass to the client transport.

The client transport converts the messages to JSON (for the Java client, this is done by a JSONContext.Client instance, see Section 8.3.1, “JSONContext API”), establishes the conduit with the server transport and then sends the JSON string over the conduit, as the payload of a transport-specific envelope (for example, an HTTP request or a WebSocket message).

The envelope travels to the server, where the server transport receives it. The server transport converts the messages from the JSON format back to message objects (through a JSONContext.Server instance, see Section 8.3.1, “JSONContext API”), then passes them to the BayeuxServer instance for processing.

The BayeuxServer processes each message in the following steps:

The round trip from client to server back to client is now complete.

When Bayeux messages is received by the server, a thread is allocated to handle the messages, and server-side listeners are invoked in this thread. The CometD implementation does not spawn new threads to call server-side listeners; in this way the threading model is kept simple and very similar to the Servlet threading model.

This simple threading model implies that if a server-side listener takes a long time to process the message and to return control to the implementation, then the implementation cannot process the next messages that may arrive, most often halting the whole server processing.

This is due to the fact that a Bayeux client uses a limited number of connections to interact with the server. If a message sent to one connection takes a long time to be processed on the server, the client may send additional messages on that connection, but those will not be processed until the previous message processing ends.

It is therefore very important that if the application knows that a message may trigger a time consuming task (for example a database query), it does so in a separate thread.

Services (see Section 8.2.2, “Using Services”) are an easy way to setup server-side listeners but share the same threading model with normal server-side listeners: if they need to perform time consuming tasks, they need to do so in a separate thread, for example:

@Service
public class MyService
{
    @Inject
    private BayeuxServer bayeuxServer;
    @Session
    private LocalSession localSession;

    @Listener("/service/query")
    public void processQuery(final ServerSession remoteSession, final ServerMessage message)
    {
        new Thread()
        {
            public void run()
            {
                Map<String, Object> data = performTimeConsumingTask(message);

                // Send data to client once the time consuming task is finished
                remoteSession.deliver(localSession, message.getChannel(), responseData, null);
            }
        }.start();
    }
}

Now that you know that applications interact with CometD through listeners, and how both the client and the server process messages, you need to know what an application should do to interact with messages to perform its business logic.

public class MySecurityPolicy extends DefaultSecurityPolicy
{
    public boolean canHandshake(BayeuxServer server, ServerSession session, ServerMessage message)
    {
        boolean authenticated = authenticate(session, message);

        if (!authenticated)
        {
            ServerMessage.Mutable reply = message.getAssociated();
            // Here you can customize the reply
        }

        return authenticated;
    }
}

@Service
public class PrivateMessageService
{
    @Session
    private ServerSession session;

    @Listener("/service/private")
    public void handlePrivateMessage(ServerSession sender, ServerMessage message)
    {
        // Retrieve the userId from the message
        String userId = message.get("targetUserId");

        // Use the mapping established during handshake to
        // retrieve the ServerSession for a given userId
        ServerSession recipient = findServerSessionFromUserId(userId);

        // Deliver the message to the other peer
        recipient.deliver(session, message.getChannel(), message.getData(), null);
    }
}

public class ExternalEventBroadcaster
{
    private final BayeuxServer bayeuxServer;

    public ExternalEventBroadcaster(BayeuxServer bayeuxServer)
    {
        this.bayeuxServer = bayeuxServer;

        // Create a local session that will act as the "sender"
        this.session = bayeuxServer.newLocalSession("external");
        this.session.handshake();
    }

    public void onExternalEvent(ExternalEvent event)
    {
        // Retrieve the channel to broadcast to, for example
        // based on the "type" property of the external event
        ServerChannel channel = this.bayeuxServer.getChannel("/events/" + event.getType());
        if (channel != null)
        {
            // Create the data to broadcast by converting the external event
            Map<String, Object> data = convertExternalEvent(event);

            // Broadcast the data
            channel.publish(this.session, data, null);
        }
    }
}

A client communicates with the server by exchanging Bayeux messages.

The Bayeux protocol requires that the first message a new client sends be a handshake message (a message sent on /meta/handshake channel). On the server, if the processing of the incoming handshake message is successful, BayeuxServer creates the server-side half-object instance (ServerSession) that represents the client that initiated the handshake. When the processing of the handshake completes, the server sends back a handshake reply to the client.

The client processes the handshake reply, and if it is successful, starts – under the covers – a heartbeat mechanism with the server, by exchanging connect messages (a message sent on a /meta/connect channel). The details of this heartbeat mechanism depend on the client transport used, but can be seen as the client sending a connect message and expecting a reply after some time (when using HTTP transports, the heartbeat mechanism is also known as long-polling). The heartbeat mechanism allows a client to detect if the server is gone (the client does not receive the connect message reply from the server), and allows the server to detect if the client is gone (the server does not receive the connect message request from the client).

Connect messages continue to flow between client and server until either side decides to disconnect by sending a disconnect message (a message sent on the /meta/disconnect channel).

While connected to the server, a client can subscribe to channels by sending a subscribe message (a message sent on a /meta/subscribe channel). Likewise, a client can unsubscribe from a channel by sending an unsubscribe message (a message sent on a /meta/unsubscribe channel). A client can publish messages containing application-specific data at any time while it is connected, and to any broadcast channel (even if it is not subscribed to that channel).

Sometimes there is the need to connect to multiple Bayeux servers. The default cometd object available as dojox.cometd or $.cometd can only be configured to connect to one server.

However, it is easy to create other cometd objects. In Dojo, there is a dojox.Cometd (note the capital 'C' of Cometd) constructor function that can be used to create new cometd objects. In jQuery, there is an equivalent $.Cometd constructor function. It can be used in this way:

// Dojo style
var cometd1 = dojox.cometd; // The default cometd object
var cometd2 = new dojox.Cometd(); // A second cometd object

// jQuery style
var cometd1 = $.cometd; // The default cometd object
var cometd2 = new $.Cometd(); // A second cometd object

// Configure and handshake
cometd1.init('http://host1:8080/cometd');
cometd2.init('http://host2:9090/cometd');

Note how the two cometd objects are initialized with different URLs.

// Dojo style
var cometd1 = dojox.cometd; // The default cometd object
var cometd2 = new dojox.Cometd(); // A second cometd object

// jQuery style
var cometd1 = $.cometd; // The default cometd object
var cometd2 = new $.Cometd(); // A second cometd object

// Configure extensions for the second object
cometd2.registerExtension('ack', new org.cometd.AckExtension());
cometd2.registerExtension('timestamp', new org.cometd.TimeStampExtension());
cometd2.registerExtension('timesync', new org.cometd.TimeSyncExtension());
cometd2.registerExtension('reload', new org.cometd.ReloadExtension());

It is not possible to subscribe to meta channels: the server replies with an error message. It is possible to listen to meta channels (see Section 7.3.4, “Subscribers versus Listeners” for an explanation of the difference between subscribers and listeners). You cannot (and it makes no sense to) publish messages to meta channels: only the Bayeux protocol implementation creates and sends messages on meta channels. Meta channels are useful on the client to listen for error messages like handshake errors (for example, because the client did not provide the correct credentials) or network errors (for example, to know when the connection with the server has broken or when it has been re-established).

Service channels are used in the case of request/response style of communication between client and server (as opposed to the publish/subscribe style of communication on broadcast channels). While subscribing to service channels yields no errors, this is a no-operation for the server: the server ignores the subscription request. It is possible to publish to service channels, with the semantic of a communication between a specific client (the one that is publishing the message on the service channel) and the server. Service channels are useful to implement, for example, private chat messages: in a chat with userA, userB and userC, userA can publish a private message to userC (without userB knowing about it) using service channels.

Broadcast channels have the semantic of a messaging topic and are used in the case of publish/subscribe style of communication. Usually, it is possible to subscribe to broadcast channels and to publish to broadcast channels; this can only be forbidden using a security policy on the Bayeux server (see Section 8.2.3, “Authorization”) or by using authorizers (see Section 8.2.4, “Server Channel Authorizers”). Broadcast channels are useful to broadcast messages to all subscribed clients, for example in case of a stock price change.

The JavaScript CometD API has two APIs to work with channel subscriptions:

The addListener() method:

The subscribe() method:

If you want to be certain that the server received your subscription request (or not), you can either register a /meta/subscribe listener, or pass a callback function to subscribe():

cometd.subscribe('/foo', function(message) { ... }, function(subscribeReply)
{
    if (subscribeReply.successful)
    {
        // The server successfully subscribed this client to the "/foo" channel.
    }
});

Remember that subscriptions may fail on the server (for example, the client does not have enough permissions to subscribe) or on the client (for example, there is a network failure). In both cases /meta/subscribe listener, or the callback function, will be invoked with an unsuccessful subscribe message reply.

You can pass additional information to the subscribe message by passing an additional object to the subscribe() function:

var additional = {
    com.acme.priority: 10
};
cometd.subscribe('/foo', function(message) { ... }, additional, function(subscribeReply)
{
    // Your logic here.
});

Remember that the subscribe message is sent to the server only when subscribing to a channel for the first time. Additional subscriptions to the same channel will not result in a message being sent to the server. See also Section 7.2, “Handshaking” for further information about passing additional objects.

Both addListener() and subscribe() return a subscription object that must be passed to, respectively, removeListener() and unsubscribe():

// Some initialization code
var subscription1 = cometd.addListener('/meta/connect', function() { ... });
var subscription2 = cometd.subscribe('/foo/bar/', function() { ... });

// Some de-initialization code
cometd.unsubscribe(subscription2);
cometd.removeListener(subscription1);

Function unsubscribe() can too take an additional object and a callback function as parameters:

// Some initialization code
var subscription1 = cometd.subscribe('/foo/bar/', function() { ... });

// Some de-initialization code
var additional = {
    com.acme.discard: true
}
cometd.unsubscribe(subscription1, additional, function(unsubscribeReply)
{
    // Your logic here.
});

Similarly to subscribe(), the unsubscribe message is sent to the server only when unsubscribing from a channel for the last time. See also Section 7.2, “Handshaking” for further information about passing additional objects.

The major difference between listeners and subscribers is that subscribers are automatically removed upon re-handshake, while listeners are not modified by a re-handshake. When a client subscribes to a channel, the server maintains a client-specific server-side subscription state. If the server requires a re-handshake, it means that it lost the state for that client, and therefore also the server-side subscription state. In order to maintain the client-side state consistent with that of the server, subscriptions - but not listeners - are automatically removed upon re-handshake.

A good place in the code to perform subscriptions is in a /meta/handshake function. Since /meta/handshake listeners are invoked in both explicit handshakes the client performs and in re-handshakes the server triggers, it is guaranteed that your subscriptions are always performed properly and kept consistent with the server state.

Equivalently, a callback function passed to the handshake method behaves exactly like a /meta/handshake listener, and therefore can be used to perform subscriptions.

Applications do not need to unsubscribe in case of re-handshake; the CometD library takes care of removing all subscriptions upon re-handshake, so that when the /meta/handshake function executes again the subscriptions are correctly restored (and not duplicated).

For the same reason, you should never add listeners inside a /meta/handshake function, because this will add another listener without removing the previous one, resulting in multiple notifications of the same messages.

var _reportListener;
cometd.addListener('/meta/handshake', function(message)
{
    // Only subscribe if the handshake is successful
    if (message.successful)
    {
        // Batch all subscriptions together
        cometd.batch(function()
        {
            // Correct to subscribe to broadcast channels
            cometd.subscribe('/members', function(m) { ... });

            // Correct to subscribe to service channels
            cometd.subscribe('/service/status', function(m) { ... });

            // Messy to add listeners after removal, prefer using cometd.subscribe(...)
            if (_reportListener)
            {
                cometd.removeListener(_reportListener);
                _reportListener = cometd.addListener('/service/report', function(m) { ... });
            }

            // Wrong to add listeners without removal
            cometd.addListener('/service/notification', function(m) { ... });
        });
    }
});

In cases where the Bayeux server is not reachable (due to network failures or because the server crashed), subscribe() and unsubscribe() behave as follows:

Often times, applications need to perform dynamic subscriptions and unsubscriptions, for example when a user clicks on a user interface element, you want to subscribe to a certain channel. In this case the subscription object returned upon subscription is stored to be able to dynamically unsubscribe from the channel upon user demand:

var _subscription;
function Controller()
{
    this.dynamicSubscribe = function()
    {
       _subscription = cometd.subscribe('/dynamic', this.onEvent);
    };

    this.onEvent = function(message)
    {
        ...
    };

    this.dynamicUnsubscribe = function()
    {
        if (_subscription)
        {
            cometd.unsubscribe(_subscription);
            _subscription = undefined;
        }
    }
}

In case of a re-handshake, dynamic subscriptions are cleared (like any other subscription) and the application needs to figure out which dynamic subscription must be performed again. This information is already known to CometD at the moment cometd.subscribe(...) was called (above in function dynamicSubscribe()), so applications can just call resubscribe() using the subscription object obtained from subscribe():

cometd.addListener('/meta/handshake', function(message)
{
    if (message.successful)
    {
        cometd.batch(function()
        {
            // Static subscription, no need to remember the subscription handle
            cometd.subscribe('/static', staticFunction);

            // Dynamic re-subscription
            if (_subscription)
            {
                _subscription = cometd.resubscribe(_subscription);
            }
        });
    }
});

If a listener or subscriber function throws an exception (for example, calls a method on an undefined object), the error message is logged at level "debug". However, there is a way to intercept these errors by defining the global listener exception handler that is invoked every time a listener or subscriber throws an exception:

cometd.onListenerException = function(exception, subscriptionHandle, isListener, message)
{
    // Uh-oh, something went wrong, disable this listener/subscriber
    // Object "this" points to the CometD object
    if (isListener)
        this.removeListener(subscriptionHandle);
    else
        this.unsubscribe(subscriptionHandle);
}

It is possible to send messages to the server from the listener exception handler. If the listener exception handler itself throws an exception, this exception is logged at level "info" and the CometD implementation does not break. Notice that a similar mechanism exists for extensions, see Chapter 9, Extensions.

It is possible to subscribe to several channels simultaneously using wildcards:

cometd.subscribe("/chatrooms/*", function(message) { ... });

A single asterisk has the meaning of matching a single channel segment; in the example above it matches channels /chatrooms/12 and /chatrooms/15, but not /chatrooms/12/upload. To match multiple channel segments, use the double asterisk:

cometd.subscribe("/events/**", function(message) { ... });

With the double asterisk, the channels /events/stock/FOO and /events/forex/EUR match, as well as /events/feed and /events/feed/2009/08/03.

The wildcard mechanism works also for listeners, so it is possible to listen to all meta channels as follows:

cometd.addListener("/meta/*", function(message) { ... });

By default, subscriptions to the global wildcards /* and /** result in an error, but you can change this behavior by specifying a custom security policy on the Bayeux server.

These are the meta channels available in the JavaScript CometD implementation:

Each meta channel is notified when the JavaScript CometD implementation handles the correspondent Bayeux message. The /meta/unsuccessful channel is notified in case of any failure.

By far the most interesting meta channel to subscribe to is /meta/connect because it gives the status of the current connection with the Bayeux server. In combination with /meta/disconnect, you can use it, for example, to display a green connected icon or a red disconnected icon on the page, depending on the connection status with the Bayeux server.

Here is a common pattern using the /meta/connect and /meta/disconnect channels:

var _connected = false;

cometd.addListener('/meta/connect', function(message)
{
    if (cometd.isDisconnected())
    {
        return;
    }

    var wasConnected = _connected;
    _connected = message.successful;
    if (!wasConnected && _connected)
    {
        // Reconnected
    }
    else if (wasConnected && !_connected)
    {
        // Disconnected
    }
});

cometd.addListener('/meta/disconnect', function(message)
{
    if (message.successful)
    {
        _connected = false;
    }
}

One small caveat with the /meta/connect channel is that /meta/connect is also used for polling the server. Therefore, if a disconnect is issued during an active poll, the server returns the active poll and this triggers the /meta/connect listener. The initial check on the status verifies that is not the case before executing the connection logic.

Another interesting use of meta channels is when there is an authentication step during the handshake. In this case the registration to the /meta/handshake channel can give details about, for example, authentication failures.

In case of temporary network failures, the client is notified through the /meta/connect channel (see Section 7.3, “Subscribing and Unsubscribing” about meta channels) with messages that have the successful field set to false (see also the archetypes in Chapter 5, Primer as an example). However, the Bayeux server might be able to keep the client's state, and when the network resumes the Bayeux server might behave as if nothing happened. The client in this case just re-establishes the long poll, but any message the client publishes during the network failure is not automatically re-sent (though it is possible to be notified, through the /meta/publish channel or better yet through callback functions, of the failed publishes).

If the network failure is long enough, the Bayeux server times out the lost client, and deletes the state associated with it. The same happens when the Bayeux server crashes (except that the state of all clients is lost). In this case, the reconnection mechanism on the client performs the following steps:

If you register meta channels listener, or if you use callback functions, be aware of these steps, since a reconnection might involve more than one message exchange with the server.

The long-polling transport is the default transport if the browser and the server do not support WebSockets. This transport is used when the communication with the Bayeux server happens on the same domain, and also in the cross-domain mode for recent browsers, such as Firefox 3.5+ (see Section 7.7.5, “The cross-domain Mode”). The data is sent to the server by means of a POST request with Content-Type application/json via a plain XMLHttpRequest call.

The callback-polling transport is used when the communication with the Bayeux server happens on a different domain and when the cross-domain mode is not supported (see Section 7.7.5, “The cross-domain Mode”).

It is well known that XMLHttpRequest calls have restrictions when the invocation is directed to a domain different from the one to which the script has been downloaded (but, again, see Section 7.7.5, “The cross-domain Mode” for an alternative solution). To overcome XMLHttpRequest restrictions, this transport uses the JSONP script injection, which injects a <script> element whose src attribute points to the Bayeux server. The browser notices the script element injection and performs a GET request to the specified source URL. The Bayeux server is aware that this is a JSONP request and replies with a JavaScript function that the browser then executes, (and calls back into the JavaScript CometD implementation).

There are three main drawbacks in using this transport:

The websocket transport is available if the browser and the server support WebSocket (Jetty 7 and later support the WebSocket protocol). The WebSocket protocol is designed to be the bidirectional communication protocol for the web, so it is a natural fit in the CometD project.

Since browser support has greatly improved, the websocket transport is enabled by default since CometD 2.4.0, but you can easily disable it if you experience problems.

The easiest way of enabling/disabling the websocket transport is to set a boolean variable before performing the initial CometD handshake:

var cometd = dojox.cometd; // Dojo style
var cometd = $.cometd; // jQuery style

// Disable the websocket transport
cometd.websocketEnabled = false;

// Initial handshake
cometd.init('http://localhost:8080/cometd');

An alternative way of disabling the websocket transport is to unregister its transport, see Section 7.7.4, “Unregistering Transports”.

CometD JavaScript transports are added in the JavaScript toolkit bindings for the CometD JavaScript library.

The CometD JavaScript API allows you to unregister transports, and this can be useful to force the use of only one transport (for example, for testing purposes), or to disable certain transports that might be unreliable. For example, it is possible to unregister the WebSocket transport by unregistering it with the following code:

var cometd = dojox.cometd; // Dojo style
var cometd = $.cometd; // jQuery style

cometd.unregisterTransport('websocket');

Firefox 3.5 introduced the capability for XMLHttpRequest calls to be performed towards a different domain (see HTTP Access Control). The JavaScript CometD implementation also supports this, with no configuration necessary on the client (if the browser supports XMLHttpRequest cross-domain calls, CometD uses them) and with a bit of configuration for the server. See the Jetty Cross Origin Filter documentation for the server configuration.

To use the cross-domain mode, you need:

With this setup, even when the communication with the Bayeux server is cross-domain, CometD uses the long-polling transport, avoiding the drawbacks of the callback-polling transport.

To initiate the communication with the Bayeux server, you need to call:

BayeuxClient.handshake()

The following is a typical use:

// Create (and eventually set up) Jetty's HttpClient:
HttpClient httpClient = new HttpClient();
// Here set up Jetty's HttpClient, for example:
// httpClient.setMaxConnectionsPerAddress(2);
httpClient.start();

// Prepare the transport
Map<String, Object> options = new HashMap<String, Object>();
ClientTransport transport = new LongPollingTransport(options, httpClient);

// Create the BayeuxClient
ClientSession client = new BayeuxClient("http://localhost:8080/cometd", transport);

// Here set up the BayeuxClient, for example:
// client.getChannel(Channel.META_CONNECT).addListener(new ClientSessionChannel.MessageListener() { ... });

BayeuxClient must be instantiated passing the absolute URL (and therefore including the scheme, host, optionally the port and the path) of the Bayeux server. The scheme of the URL must always be either "http" or "https". The CometD Java Client implementation will transparently take care of converting the scheme to "ws" or "wss" in case of usage of the WebSocket protocol.

When BayeuxClient.handshake() is called, the BayeuxClient performs the handshake with the Bayeux server and establishes the long poll connection asynchronously.

To verify that the handshake is successful, you can pass a callback MessageListener to BayeuxClient.handshake():

ClientTransport transport = ...
ClientSession client = new BayeuxClient("http://localhost:8080/cometd", transport);
client.handshake(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        if (message.isSuccessful())
        {
            // Here handshake is successful
        }
    }
});

An alternative, equivalent, way is to add a MessageListener before calling BayeuxClient.handshake():

ClientTransport transport = ...
ClientSession client = new BayeuxClient("http://localhost:8080/cometd", transport);
client.getChannel(Channel.META_HANDSHAKE).addListener(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        if (message.isSuccessful())
        {
            // Here handshake is successful
        }
    }
});
client.handshake();

Another alternative is to use the built-in synchronous features of the BayeuxClient and wait for the handshake to complete:

ClientTransport transport = ...
BayeuxClient client = new BayeuxClient("http://localhost:8080/cometd", transport);
client.handshake();
boolean handshaken = client.waitFor(1000, BayeuxClient.State.CONNECTED);
if (handshaken)
{
    // Here handshake is successful
}

The BayeuxClient.waitFor() method waits the given timeout (in milliseconds) for the BayeuxClient to reach the given state, and returns true if the state is reached before the timeout expires.

The following sections provide information about subscribing and unsubscribing to channels.

public class Example
{
    private static final String CHANNEL = "/foo";
    private final ClientSessionChannel.MessageListener fooListener = new FooListener();

    public void attach()
    {
        ClientTransport transport = ...
        ClientSession client = new BayeuxClient("http://localhost:8080/cometd", transport);
        client.handshake();
        boolean handshaken = client.waitFor(1000, BayeuxClient.State.CONNECTED);
        if (handshaken)
        {
            client.getChannel(CHANNEL).subscribe(fooListener);
        }
    }

    private static class FooListener implements ClientSessionChannel.MessageListener
    {
        public void onMessage(ClientSessionChannel channel, Message message)
        {
            // Here we received a message on the channel
        }
    }
}

public class Example
{
    ...
    public void detach()
    {
        client.getChannel(CHANNEL).unsubscribe(fooListener);
    }
}

final BayeuxClient client = ...;
final ClientSessionChannel.MessageListener messageHandler = ...;
client.handshake(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        if (message.isSuccessful())
        {
            // Subscribe
            client.getChannel("/foo").subscribe(messageHandler, new ClientSessionChannel.MessageListener()
            {
                public void onMessage(ClientSessionChannel channel, Message message)
                {
                    if (message.isSuccessful())
                    {
                        // Subscription successful.
                    }
                }
            }
        }
    }
});

public class Example
{
    public void init()
    {
        ClientSession client = ...;
        client.getChannel(Channel.META_HANDSHAKE).addListener(new ClientSessionChannel.MessageListener()
        {
            public void onMessage(ClientSessionChannel channel, Message message)
            {
                // Here we received a handshake response message
            }
        });
    }
}

The following is a typical example of publishing a message on a channel:

ClientTransport transport = ...
ClientSession client = new BayeuxClient("http://localhost:8080/cometd", transport);
client.handshake();

Map<String, Object> data = new HashMap<String, Object>();
// Fill in the data

client.getChannel("/game/table/1").publish(data);

Publishing data on a channel is an asynchronous operation.

When the message you published arrives to the server, the server replies to the client with a publish acknowledgment; this allows clients to be sure that the message reached the server. The publish acknowledgment arrives on the same channel the message was published to, with the same message id, with a successful field. If the message publish fails for any reason, for example because server cannot be reached, then a publish failure will be emitted, similarly to publish acknowledgments.

In order to be notified of publish acknowledgments or failures, you can use this variant of the publish() method:

Map<String, Object> data = new HashMap<String, Object>();
// Fill in the data

client.getChannel("/game/table/1").publish(data, new ClientSessionChannel.MessageListener()
{
    @Override
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        if (message.isSuccessful())
        {
            // The message reached the server
        }
    }
});

Message batching is also available:

final ClientSession client = ...;
client.handshake();

client.batch (new Runnable()
{
    public void run()
    {
        Map<String, Object> data1 = new HashMap<String, Object>();
        // Fill in the data1 map object
        client.getChannel("/game/table/1").publish(data1);

        Map<String, Object> data2 = new HashMap<String, Object>();
        // Fill in the data2 map object<
        client.getChannel("/game/chat/1").publish(data2);
    }
});

Disconnecting is straightforward:

BayeuxClient.disconnect();

Like the other APIs, you can pass a callback MessageListener to be notified that the server received and processed the disconnect request:

BayeuxClient client = ...;
client.disconnect(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        if (message.isSuccessful())
        {
            // Server processed the disconnect request.
        }
    }
});

Alternatively, you can wait for the disconnect to complete:

BayeuxClient client = ...;
client.disconnect();
client.waitFor(1000, BayeuxClient.State.DISCONNECTED);

You can configure org.cometd.client.BayeuxClient class to use multiple transports. It currently supports the long-polling transport (that in turn depends on Jetty's asynchronous HttpClient) and the websocket transport (that in turn depends on Jetty's asynchronous WebSocket client.

You should configure BayeuxClient with the websocket transport before the long-polling transport, so that BayeuxClient can fall back to the long-polling if the websocket transport fails. You do so by listing the WebSocket transport before the HTTP transport on the BayeuxClient constructor, for example:

// Prepare the WebSocket transport
WebSocketClientFactory wsFactory = new WebSocketClientFactory();
wsFactory.start();
ClientTransport wsTransport = new WebSocketTransport(null, wsFactory, null);

// Prepare the HTTP transport
HttpClient httpClient = new HttpClient();
httpClient.start();
ClientTransport  httpTransport = new LongPollingTransport(null, httpClient);

// Configure the BayeuxClient, with the websocket transport listed before the http transport
BayeuxClient client = new BayeuxClient("http://localhost:8080/cometd", wsTransport, httpTransport);

// Handshake
client.handshake();

You can specify BayeuxServer and server transport parameters in web.xml as init parameters of the org.cometd.server.CometdServlet. If the CometD servlet creates the BayeuxServer instance, the servlet init parameters are passed to the BayeuxServer instance, which in turn configures the server transports.

If you followed Chapter 5, Primer, Maven has configured the web.xml file for you; here are details its configuration, a sample web.xml:

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <init-param>
            <param-name>timeout</param-name>
            <param-value>60000</param-value>
        </init-param>
        <init-param>
            <param-name>logLevel</param-name>
            <param-value>3</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

</web-app>

You must define and map the org.cometd.server.CometdServlet in web.xml to enable the server to interpret the Bayeux protocol. It is normally mapped on /cometd/*, but you can change the mapping url-pattern if you prefer.

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <init-param>
            <param-name>timeout</param-name>
            <param-value>60000</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <filter>
        <filter-name>cross-origin</filter-name>
        <filter-class>org.eclipse.jetty.servlets.CrossOriginFilter</filter-class>
    </filter>
    <filter-mapping>
        <filter-name>cross-origin</filter-name>
        <url-pattern>/cometd/*</url-pattern>
    </filter-mapping>

</web-app>

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <filter>
        <filter-name>continuation</filter-name>
        <filter-class>org.eclipse.jetty.continuation.ContinuationFilter</filter-class>
    </filter>
    <filter-mapping>
        <filter-name>continuation</filter-name>
        <url-pattern>/cometd/*</url-pattern>
    </filter-mapping>

</web-app>

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
         version="3.0">                                                                  (1)

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <async-supported>true</async-supported>                                          (2)
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <filter>
        <filter-name>cross-origin</filter-name>
        <filter-class>org.eclipse.jetty.servlets.CrossOriginFilter</filter-class>
        <async-supported>true</async-supported>                                          (2)
    </filter>
    <filter-mapping>
        <filter-name>cross-origin</filter-name>
        <url-pattern>/cometd/*</url-pattern>
    </filter-mapping>

</web-app>

IllegalStateException: the servlet does not support async operations for this request

BayeuxServer bayeuxServer = ...;
boolean created = bayeuxServer.createIfAbsent("/my/channel", new ServerChannel.Initializer()
{
    public void configureChannel(ConfigurableServerChannel channel)
    {
        // Here configure the channel
    }
});

BayeuxServer bayeuxServer = ...;
boolean created = bayeuxServer.createIfAbsent("/my/channel", new ServerChannel.Initializer()
{
    public void configureChannel(ConfigurableServerChannel channel)
    {
        channel.setPersistent(true);
    }
});

BayeuxServer bayeuxServer = ...;
String channelName = "/my/channel";
bayeuxServer.createIfAbsent(channelName, new ServerChannel.Initializer()
{
    public void configureChannel(ConfigurableServerChannel channel)
    {
        channel.setPersistent(true);
    }
});
ServerChannel channel = bayeuxServer.getChannel(channelName);

BayeuxServer bayeuxServer = ...;
String channelName = "/my/channel";

// Wrong, channel not marked as persistent, but used later
bayeuxServer.createIfAbsent(channelName);

// Other application code here

ServerChannel channel = bayeuxServer.getChannel(channelName);
channel.publish(...); // May throw NullPointerException

A CometD service is a Java class that allows a developer to specify the code to run when Bayeux channels receive Bayeux messages. When a message arrives on a channel to which the service instance subscribes, CometD invokes a callback method to execute user-specific code.

CometD 2 services can be of two kinds:

You can also integrate CometD 2 services with the Spring Framework as explained in Section 8.2.2.4, “Services Integration with Spring”.

The following sections present details about Java Server Services.

public class EchoService extends AbstractService                                  (1)
{
    public EchoService(BayeuxServer bayeuxServer)                                 (2)
    {
        super(bayeuxServer, "echo");                                              (3)
        addService("/echo", "processEcho");                                       (4)
    }

    public void processEcho(ServerSession remote, Map<String, Object> data)       (5)
    {
        remote.deliver(getServerSession(), "/echo", data, null);                  (6)
    }
}

// Obtains the remote session object and the message object
public void processEcho(ServerSession remote, Message message)

// Obtains the remote session object and the message's data object
// (additional message information, such as the channel or the id is lost)
public void processEcho(ServerSession remote, Map<String, Object> data)

// Obtains the remote session object, the channel name, the message object and the message id
public void processEcho(ServerSession remote, String channelName, Message message, String messageId)

// Obtains the remote session object, the channel name, the message's data object and the message id
public void processEcho(ServerSession remote, String channelName, Map<String, Object> data, String messageId)

// Obtains the remote session object and a custom message's data object
public void processEcho(ServerSession remote, EchoInfo data)

public class BaseballTeamService extends AbstractService
{
    public BaseballTeamService(BayeuxServer bayeux)
    {
        super(bayeux, "baseballTeam");
        addService("/baseball/team/*", "processBaseballTeam");
    }

    public void processBaseballTeam(ServerSession remote, String channelName, Map<String, Object> data, String messageId)
    {
        // Upon receiving a message on channel /baseball/team/*, forward to channel /events/baseball/team/*
        getBayeux().getChannel("/events" + channelName).publish(getServerSession(), data, null);
    }
}

public class GameService extends AbstractService
{
    public GameService(BayeuxServer bayeux)
    {
        super(bayeux, "game");
        addService("/service/game/*", "processGameCommand");
        addService("/game/event", "processGameEvent");
    }

    public void processGameCommand(ServerSession remote, GameCommand command)
    {
        switch (command.getType())
        {
            case GAME_START:
            {
                addService("/game/" + command.getGameId(), "processGame");
                break;
            }
            case GAME_END:
            {
                removeService("/game/" + command.getGameId());
                break;
            }
            ...
        }
    }

    public void processGameEvent(ServerSession remote, GameEvent event)
    {
        ...
    }
}

@org.cometd.annotation.Service("echoService")
public class EchoService
{
    @javax.inject.Inject
    private BayeuxServer bayeux;
}

public class EchoService extends AbstractService
{
    public EchoService(BayeuxServer bayeux)
    {
        super(bayeux, "echoService");
    }
}

@Service("echoService")
public class EchoService
{
    @Inject
    private BayeuxServer bayeux;

    @Configure("/echo")
    public void configure(ConfigurableServerChannel channel)
    {
        channel.setLazy(true);
        channel.addAuthorizer(GrantAuthorizer.GRANT_PUBLISH);
    }
}

@Service("echoService")
public class EchoService
{
    @Inject
    private BayeuxServer bayeux;

    @org.cometd.annotation.Session
    private LocalSession localSession;

    @org.cometd.annotation.Session
    private ServerSession serverSession;
}

@Service("echoService")
public class EchoService
{
    @Inject
    private BayeuxServer bayeux;
    @Session
    private ServerSession serverSession;

    @org.cometd.annotation.Listener("/echo")
    public void echo(ServerSession remote, ServerMessage.Mutable message)
    {
        String channel = message.getChannel();
        Object data = message.getData();
        remote.deliver(serverSession, channel, data, null);
    }
}

public class EchoService extends AbstractService
{
    public EchoService(BayeuxServer bayeux)
    {
        super(bayeux, "echoService");
        addService("/echo", "echo");
    }

    public void echo(ServerSession remote, ServerMessage.Mutable message)
    {
        String channel = message.getChannel();
        Object data = message.getData();
        remote.deliver(getServerSession(), channel, data, null);
    }
}

@Service("echoService")
public class EchoService
{
    @Inject
    private BayeuxServer bayeux;
    @Session
    private ServerSession serverSession;

    @org.cometd.annotation.Subscription("/echo")
    public void echo(Message message)
    {
        System.out.println("Echo service published " + message);
    }
}

BayeuxServer bayeux = ...;

// Create the ServerAnnotationProcessor
ServerAnnotationProcessor processor = new ServerAnnotationProcessor(bayeux);

// Create the service instance
EchoService service = new EchoService();

// Process the annotated service
processor.process(service);

@Service
public class Service
{
    @Session
    private ClientSession bayeuxClient;

    @Listener(Channel.META_CONNECT)
    public void metaConnect(Message connect)
    {
        // Connect handling...
    }

    @Subscription("/foo")
    public void foo(Message message)
    {
        // Message handling...
    }
}

ClientSession bayeuxClient = ...;

bayeuxClient.getChannel(Channel.META_CONNECT).addListener(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        // Connect handling...
    }
});

bayeuxClient.handshake();
bayeuxClient.waitFor(1000, BayeuxClient.State.CONNECTED);

bayeuxClient.getChannel("/foo").subscribe(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        // Message handling...
    }
});

@Service
public class Service
{
    @org.cometd.annotation.Session
    private ClientSession bayeuxClient;
}

@Service
public class Service
{
    @Listener(Channel.META_CONNECT)
    public void metaConnect(Message connect)
    {
        // Connect handling...
    }
}

bayeuxClient.getChannel(Channel.META_CONNECT).addListener(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        // Connect handling...
    }
});

@Service
public class Service
{
    @Listener("/foo/*")
    public void foos(Message message)
    {
       // Message handling...
    }
}

bayeuxClient.getChannel("/foo/*").subscribe(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        // Message handling...
    }
});

ClientSession bayeuxClient = ...;

// Create the ClientAnnotationProcessor
ClientAnnotationProcessor processor = new ClientAnnotationProcessor(bayeuxClient);

// Create the service instance
Service service = new Service();

// Process the annotated service
processor.process(service);

bayeuxClient.handshake();

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <servlet>
        <servlet-name>configuration</servlet-name>
        <servlet-class>com.acme.cometd.ConfigurationServlet</servlet-class>
        <load-on-startup>2</load-on-startup>
    </servlet>

</web-app>

public class ConfigurationServlet extends GenericServlet
{
    public void init() throws ServletException
    {
        // Grab the Bayeux object
        BayeuxServer bayeux = (BayeuxServer)getServletContext().getAttribute(BayeuxServer.ATTRIBUTE);
        new EchoService(bayeux);
        // Create other services here

        // This is also the place where you can configure the Bayeux object
        // by adding extensions or specifying a SecurityPolicy
    }

    public void service(ServletRequest request, ServletResponse response) throws ServletException, IOException
    {
        throw new ServletException();
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <listener>
        <listener-class>com.acme.cometd.BayeuxInitializer</listener-class>
    </listener>

</web-app>

public class BayeuxInitializer implements ServletContextAttributeListener
{
    public void attributeAdded(ServletContextAttributeEvent event)
    {
        if (Bayeux.ATTRIBUTE.equals(event.getName()))
        {
            // Grab the Bayeux object
            BayeuxServer bayeux = (BayeuxServer)event.getValue();
            new EchoService(bayeux);
            // Create other services here

            // This is also the place where you can configure the Bayeux object
            // by adding extensions or specifying a SecurityPolicy
        }
    }

    public void attributeRemoved(ServletContextAttributeEvent event)
    {
    }

    public void attributeReplaced(ServletContextAttributeEvent event)
    {
    }
}

public class ConfigurationServlet extends GenericServlet
{
    private final List<Object> services = new ArrayList<Object>();
    private ServerAnnotationProcessor processor;

    public void init() throws ServletException
    {
        // Grab the BayeuxServer object
        BayeuxServer bayeux = (BayeuxServer)getServletContext().getAttribute(BayeuxServer.ATTRIBUTE);

        // Create the annotation processor
        processor = new ServerAnnotationProcessor(bayeux);

        // Create your annotated service instance and process it
        Object service = new EchoService();
        processor.process(service);
        services.add(service);

        // Create other services here

        // This is also the place where you can configure the Bayeux object
        // by adding extensions or specifying a SecurityPolicy
    }

    public void destroy() throws ServletException
    {
        // Deprocess the services that have been created
        for (Object service : services)
            processor.deprocess(service);
    }

    public void service(ServletRequest request, ServletResponse response) throws ServletException, IOException
    {
        throw new ServletException();
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.java.annotation.AnnotationCometdServlet</servlet-class>
        <init-param>
            <param-name>services</param-name>
            <param-value>com.acme.cometd.FooService, com.acme.cometd.BarService</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

</web-app>

public class AnotherServlet extends HttpServlet
{
    protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
    {
        FooService service = (FooService)getServletContext().getAttribute("com.acme.cometd.FooService");
        // Use the foo service here
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>

</web-app>

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">

    <bean id="otherService" class="com.acme..." />

    <bean id="bayeux" class="org.cometd.server.BayeuxServerImpl" init-method="start" destroy-method="stop">
        <property name="options">
            <map>
                <entry key="logLevel" value="3" />
                <entry key="timeout" value="15000" />
            </map>
        </property>
    </bean>

    <bean id="echoService" class="com.acme.cometd.EchoService">
        <constructor-arg><ref local="bayeux" /></constructor-arg>
        <constructor-arg><ref local="otherService" /></constructor-arg>
    </bean>

    <bean class="org.springframework.web.context.support.ServletContextAttributeExporter">
        <property name="attributes">
            <map>
                <entry key="org.cometd.bayeux">
                    <ref local="bayeux" />
                </entry>
            </map>
        </property>
    </bean>
</beans>

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">

    <bean id="bayeux" class="org.cometd.server.BayeuxServerImpl" init-method="start" destroy-method="stop">
        <property name="transports">
            <list>
                <bean id="websocketTransport" class="org.cometd.websocket.server.WebSocketTransport">
                    <constructor-arg ref="bayeux" />
                </bean>
                <bean id="jsonTransport" class="org.cometd.server.transport.JSONTransport">
                    <constructor-arg ref="bayeux" />
                </bean>
                <bean id="jsonpTransport" class="org.cometd.server.transport.JSONPTransport">
                    <constructor-arg ref="bayeux" />
                </bean>
            </list>
        </property>
    </bean>

    ... as before ...

</beans>

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
                           http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">

    <context:component-scan base-package="com.acme..." />

</beans>

@javax.inject.Named // Tells Spring that this is a bean
@javax.inject.Singleton // Tells Spring that this is a singleton
@Service("echoService")
public class EchoService
{
    @Inject
    private BayeuxServer bayeux;
    @Session
    private ServerSession serverSession;

    @PostConstruct
    public void init()
    {
        System.out.println("Echo Service Initialized");
    }

    @Listener("/echo")
    public void echo(ServerSession remote, ServerMessage.Mutable message)
    {
        String channel = message.getChannel();
        Object data = message.getData();
        remote.deliver(serverSession, channel, data, null);
    }
}

@Component
public class Configurer implements DestructionAwareBeanPostProcessor, ServletContextAware
{
    private BayeuxServer bayeuxServer;
    private ServerAnnotationProcessor processor;

    @Inject
    private void setBayeuxServer(BayeuxServer bayeuxServer)
    {
        this.bayeuxServer = bayeuxServer;
    }

    @PostConstruct
    private void init()
    {
        this.processor = new ServerAnnotationProcessor(bayeuxServer);
    }

    public Object postProcessBeforeInitialization(Object bean, String name) throws BeansException
    {
        processor.processDependencies(bean);
        processor.processConfigurations(bean);
        processor.processCallbacks(bean);
        return bean;
    }

    public Object postProcessAfterInitialization(Object bean, String name) throws BeansException
    {
        return bean;
    }

    public void postProcessBeforeDestruction(Object bean, String name) throws BeansException
    {
        processor.deprocessCallbacks(bean);
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
    public BayeuxServer bayeuxServer()
    {
        BayeuxServerImpl bean = new BayeuxServerImpl();
        bean.setOption(BayeuxServerImpl.LOG_LEVEL, "3");
        return bean;
    }

    public void setServletContext(ServletContext servletContext)
    {
        servletContext.setAttribute(BayeuxServer.ATTRIBUTE, bayeuxServer);
    }
}

You can configure the BayeuxServer object with an org.cometd.bayeux.server.SecurityPolicy object, which allows you to control various steps of the Bayeux protocol such as handshake, subscription, and publish. By default, the BayeuxServer object has a default SecurityPolicy, that allows almost any operation.

The org.cometd.bayeux.server.SecurityPolicy has a default implementation in org.cometd.server.DefaultSecurityPolicy, that is useful as a base class to customize the SecurityPolicy (see how authentication works for an example).

The org.cometd.bayeux.server.SecurityPolicy methods are:

boolean canHandshake(BayeuxServer server, ServerSession session, ServerMessage message);

boolean canCreate(BayeuxServer server, ServerSession session, String channelId, ServerMessage message);

boolean canSubscribe(BayeuxServer server, ServerSession session, ServerChannel channel, ServerMessage messsage);

boolean canPublish(BayeuxServer server, ServerSession session, ServerChannel channel, ServerMessage messsage);

Those methods control, respectively, whether a handshake, a channel creation, a subscription to a channel or a publish to a channel are to be authorized.

The default implementation org.cometd.server.DefaultSecurityPolicy:

A typical custom SecurityPolicy may override the canHandshake(...) method to control authentication:

public class CustomSecurityPolicy extends DefaultSecurityPolicy
{
    public boolean canHandshake(BayeuxServer server, ServerSession session, ServerMessage message)
    {
        // Always allow local clients
        if (session.isLocalSession())
            return true;

        // Implement custom authentication logic
        boolean authenticated = authenticate(server, session, message);

        return authenticated;
    }
}

To understand how to install your custom SecurityPolicy on the BayeuxServer object, see how it is done in the authentication howto.

Follow also the section on Section 8.2.4, “Server Channel Authorizers” for further information about authorization.

CometD 2.1.0 introduced authorizers, a feature that allows fine-grained control of authorization on channel operations.

BayeuxServer bayeuxServer = ...;
bayeuxServer.createIfAbsent("/my/channel", new ConfigurableServerChannel.Initializer()
{
    public void configureChannel(ConfigurableServerChannel channel)
    {
        channel.addAuthorizer(GrantAuthorizer.GRANT_PUBLISH);
    }
});

public boolean canCreate   (BayeuxServer server, ServerSession session, String channelId,      ServerMessage message);
public boolean canSubscribe(BayeuxServer server, ServerSession session, ServerChannel channel, ServerMessage message);
public boolean canPublish  (BayeuxServer server, ServerSession session, ServerChannel channel, ServerMessage message);

public Result authorize(Operation operation, ChannelId channel, ServerSession session, ServerMessage message);

BayeuxServer bayeuxServer = ...;
bayeuxServer.createIfAbsent("/game/**");
ServerChannel gameStarStar = bayeuxServer.getChannel("/game/**");
gameStarStar.addAuthorizer(GrantAuthorizer.GRANT_NONE);

gameStarStar.addAuthorizer(new Authorizer()
{
    public Result authorize(Operation operation, ChannelId channel, ServerSession session, ServerMessage message)
    {
        // Always grant authorization to local clients
        if (session.isLocalSession())
            return Result.grant();

        boolean isCaptain = isCaptain(session);
        boolean isGameChannel = !channel.isWild() && new ChannelId("/game").isParentOf(channel);
        if (operation == Operation.CREATE && isCaptain && isGameChannel)
            return Result.grant();
        return Result.ignore();
    }
});

gameStarStar.addAuthorizer(GrantAuthorizer.GRANT_SUBSCRIBE);

ServerChannel gameChannel = bayeuxServer.getChannel("/game/" + gameId);
gameChannel.addAuthorizer(new Authorizer()
{
    public Result authorize(Operation operation, ChannelId channel, ServerSession session, ServerMessage message)
    {
        // Always grant authorization to local clients
        if (session.isLocalSession())
            return Result.grant();

        boolean isPlayer = isPlayer(session, channel);
        if (operation == Operation.PUBLISH && isPlayer)
            return Result.grant();
        return Result.ignore();
    }
});

/game/**  --> one authorizer that ignores everything
          --> one authorizer that grants captains authority to create games
          --> one authorizer that grants everyone the ability to watch games
/game/123 --> one authorizer that grants players the ability to play

gameStarStar.addAuthorizer(new Authorizer()
{
    public Result authorize(Operation operation, ChannelId channel, ServerSession session, ServerMessage message)
    {
        // Always grant authorization to local clients
        if (session.isLocalSession())
            return Result.grant();

        boolean isCriminalSupporter = isCriminalSupporter(session);
        if (operation == Operation.SUBSCRIBE && isCriminalSupporter)
            return Result.deny("criminal_supporter");
        return Result.ignore();
    }
});

/game/**  --> one authorizer that ignores everything
          --> one authorizer that grants captains the ability to create games
          --> one authorizer that grants everyone the ability to watch games
          --> one authorizer that denies criminal supporters the ability to watch games
/game/123 --> one authorizer that grants players the ability to play

Appendix C, The Bayeux Protocol Specification v1.0 defines two mandatory transports: long-polling and callback-polling. These two transports are automatically configured in the BayeuxServer instance, and there is no need to explicitly specify them in the configuration.

For the websocket transport things are different because there is no standard API for server-side WebSocket yet (it is under consideration for the Servlet 3.1 specification), and therefore any implementation binds you to the servlet container implementation.

To configure the websocket transport on the server, follow these instructions:

If you have configured your web application to support cross-domain HTTP calls (see Section 8.2.1.3, “Configuring the CrossOriginFilter”), you do not need to worry because the CrossOriginFilter is by default disabled for the WebSocket protocol.

Server-side components such as services (see Section 8.2.2, “Using Services”), extensions (see Chapter 9, Extensions) or authorizers (see Section 8.2.4, “Server Channel Authorizers”) may need to access contextual information, that is information provided by the Servlet environment such as request attributes, HTTP session attributes, servlet context attributes or network address information.

While the Servlet model can easily provide such information, CometD can use also non-HTTP transports such as WebSocket that may not have such information readily available. CometD abstracts the retrieval of contextual information for any transport via the org.cometd.bayeux.server.BayeuxContext class.

An instance of BayeuxContext can be obtained from the BayeuxServer instance:

BayeuxContext context = bayeuxServer.getContext();

A typical usage in a SecurityPolicy (see Section 8.2.3, “Authorization”) is the following:

public class OneClientPerAddressSecurityPolicy extends DefaultSecurityPolicy
{
    private final Set<String> addresses = new HashSet<String>();

    @Override
    public boolean canHandshake(BayeuxServer bayeuxServer, ServerSession session, ServerMessage message)
    {
        BayeuxContext context = bayeuxServer.getContext();

        // Get the remote address of the client
        final String remoteAddress = context.getRemoteAddress().getHostName();

        // Only allow clients from different remote addresses

        boolean notPresent = addresses.add(remoteAddress);

        // Avoid to leak addresses
        session.addListener(new ServerSession.RemoveListener()
        {
            public void removed(ServerSession session, boolean timeout);
            {
                addresses.remove(remoteAddress);
            }
        });

        return notPresent ? true : false;
    }
}

Refer to the Javadoc documentation for further information about methods available in BayeuxContext.

It is recommended to always call BayeuxServer.getContext() to obtain the BayeuxContext instance; it should never be cached and then reused across messages. This allows maximum portability of your code in case you're using a mix of WebSocket and HTTP transports.

Sometimes applications need to send non-urgent messages to clients; alert messages such as "email received" or "server uptime", are typical examples, but chat messages sometimes also fit this definition. While these messages need to reach the client(s) eventually, there is no need to deliver them immediately: they are queued into the ServerSession's message queue, but their delivery can be postponed to the first occasion.

In CometD, "first occasion" means whichever of the following things occurs first (see Section 8.2.1, “Configuring the Java Server” for information about configuring the following parameters):

To support this feature, CometD introduces the concepts of lazy channel and lazy messages.

An application can mark a message as lazy by calling ServerMessage.Mutable.setLazy(true) on the message instance itself, for example:

@Service
public class LazyService
{
    @Inject
    private BayeuxServer bayeuxServer;
    @Session
    private LocalSession session;

    public void receiveNewsFromExternalSystem(NewsInfo news)
    {
        ServerMessage.Mutable message = this.bayeuxServer.newMessage();
        message.setChannel("/news");
        message.setData(news);
        message.setLazy(true);
        this.bayeuxServer.getChannel("/news").publish(this.session, message);
    }
}

In this example, an external system invokes method LazyService.receiveNewsFromExternalSystem(...) every time there is news, and the service broadcasts the news to all interested clients. However, since the news need not be delivered immediately, the message is marked as lazy. Each remote client possibly receives the message at a different time: some receive it immediately (because, for example, they have other non-lazy messages to be delivered), some receive it after few milliseconds (because, for example, their /meta/connect timeout expires in a few milliseconds), while others receive it only upon the whole maxLazyInterval timeout.

In the same way you can mark a message as lazy, you can also mark server channels as lazy. Every message that is published to a lazy channel becomes a lazy message which is then queued into the ServerSession's message queue for delivery on first occasion.

You can mark server channels as lazy at any time, but it is best to configure them as lazy at creation time, for example:

@Service
public class LazyService
{
    @Inject
    private BayeuxServer bayeuxServer;
    @Session
    private LocalSession session;

    @Configure("/news")
    public void setupNewsChannel(ConfigurableServerChannel channel)
    {
        channel.setLazy(true);
    }

    public void receiveNewsFromExternalSystem(NewsInfo news)
    {
        this.bayeuxServer.getChannel("/news").publish(this.session, news, null);
    }
}

When a server channel is marked lazy, by default it has a lazy timeout specified by the global maxLazyTimeout parameter (see Section 8.2.1, “Configuring the Java Server”).

In more sophisticated cases, you may want to specify different lazy timeouts for different server channels, for example:

@Service
public class LazyService
{
    ...

    @Configure("/news")
    public void setupNewsChannel(ConfigurableServerChannel channel)
    {
        channel.setLazy(true);
    }

    @Configure("/chat")
    public void setupChatChannel(ConfigurableServerChannel channel)
    {
        channel.setLazyTimeout(2000);
    }

    ...
}

In the example above, channel /news inherits the default maxLazyTimeout (5000 ms), while the /chat channel is configured with a specific lazy timeout of 2000 ms. A server channel that has a non-zero, positive lazy timeout is automatically marked as lazy.

If a wildcard server channel such as /chat/* is marked as lazy, then all messages sent to server channels that match that wildcard server channel (such as /chat/1) will be lazy. Conversely, if a non-wildcard server channel such as /news is lazy, then all messages sent to children server channels of that non-wildcard server channel (such as /news/sport) will not be lazy.

The HTTP protocol recommends a connection limit of two connections per domain. While modern browsers are configured by default with more than two connections per domain, it is not safe to make such an assumption; thus any iframes, tabs or windows on the same browser connecting to the same host need to share two connections.

If two iframes/tabs/windows initiate a Bayeux communication, both start a long poll connect request and both connections are consumed, making it impossible to send another Bayeux request (for example, a publish) until one of the two long polls returns.

The CometD Server implements the multiple-clients advice (see the section called “multiple-clients advice field”). The server uses BAYEUX_BROWSER cookie to detect multiple CometD clients from the same browser.

If the CometD server detects multiple clients, only one long poll connection is allowed, and subsequent long poll requests do not wait for messages before returning. They return immediately with the multiple-clients field of the advice object set to true. The advice also contains an interval field set to the value of the multiSessionInterval servlet init parameter (see Section 8.2.1, “Configuring the Java Server”). This instructs the client not to send another poll until that interval has passed. The effect of this advice is that additional client connections normally poll the server with a period of multiSessionInterval. This avoids consuming both HTTP connections at the cost of some latency for the additional iframes/tabs/windows.

We recommend that the client application monitor the /meta/connect channel for multiple-clients field in the advice. If detected, the application might ask the user to close the additional tabs, or it could automatically close them, or take some other action.

Non-browser clients (or browsers with cookies disabled) must handle the BAYEUX_BROWSER cookie with the same semantic of browsers, or configure the server to allow multiple sessions even without BAYEUX_BROWSER information via the allowMultiSessionsNoBrowser servlet init parameter (see Section 8.2.1, “Configuring the Java Server”).

Server-side CometD components may be exposed as JMX MBeans and visible via monitoring tools such as JConsole or JVisualVM.

CometD MBeans build on Jetty's JMX support, and are portable across servlet containers.

<?xml version="1.0" encoding="ISO-8859-1"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <context-param>
        <param-name>org.eclipse.jetty.server.context.ManagedAttributes</param-name>
        <param-value>org.cometd.bayeux,org.cometd.oort.Oort</param-value>
    </context-param>

    <!-- The rest of your web.xml -->
</webapp>

import java.io.IOException;
import java.lang.management.ManagementFactory;
import javax.servlet.GenericServlet;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import org.cometd.bayeux.server.BayeuxServer;
import org.eclipse.jetty.jmx.MBeanContainer;

public class CometDJMXExporter extends GenericServlet
{
    private volatile MBeanContainer mbeanContainer;

    @Override
    public void init() throws ServletException
    {
        try
        {
            mbeanContainer = new MBeanContainer(ManagementFactory.getPlatformMBeanServer());
            BayeuxServer bayeuxServer = (BayeuxServer)getServletContext().getAttribute(BayeuxServer.ATTRIBUTE);
            mbeanContainer.addBean(bayeuxServer);
            // Add other components
            mbeanContainer.start();
        }
        catch (Exception x)
        {
            throw new ServletException(x);
        }
    }

    @Override
    public void service(ServletRequest servletRequest, ServletResponse servletResponse) throws ServletException, IOException
    {
        throw new ServletException();
    }

    @Override
    public void destroy()
    {
        try
        {
            mbeanContainer.stop();
        }
        catch (Exception ignored)
        {
        }
    }
}

<?xml version="1.0" encoding="ISO-8859-1"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <!-- The rest of your web.xml -->

    <servlet>
        <servlet-name>jmx-exporter</servlet-name>
        <servlet-class>com.acme.cometd.CometDJMXExporter</servlet-class>
        <!-- Make sure it's the last started -->
        <load-on-startup>2</load-on-startup>
    </servlet>

</webapp>

The CometD Java client implementation (see Section 8.1, “Client Library”) uses the JSON library to generate JSON from and to parse JSON to org.cometd.bayeux.Message instances. The JSON library class that performs this generation/parsing on the client must implement org.cometd.common.JSONContext.Client.

Similarly, on the server, a org.cometd.common.JSONContext.Server implementation generates JSON from and parses JSON to org.cometd.bayeux.server.ServerMessage instances.

HttpClient httpClient = ...;

Map<String, Object> transportOptions = new HashMap<String, Object>();

// Use the Jackson implementation
JSONContext.Client jsonContext = new JacksonJSONContextClient();
transportOptions.put(ClientTransport.JSON_CONTEXT, jsonContext);

ClientTransport transport = new LongPollingTransport(transportOptions, httpClient);

BayeuxClient client = new BayeuxClient(cometdURL, transport);

public class MyJacksonJSONContextClient extends org.cometd.common.JacksonJSONContextClient
{
    public MyJacksonJSONContextClient()
    {
        org.codehaus.jackson.map.ObjectMapper objectMapper = getObjectMapper();
        objectMapper.registerModule(new MyModule());
    }

    private class MyModule extends org.codehaus.jackson.map.module.SimpleModule
    {
        public MyModule()
        {
            // Add your custom serializers/deserializers here
            addSerializer(Foo.class, new FooSerializer());
        }
    }
}

public EchoInfo readFromConfigFile(BayeuxServer bayeuxServer) throws ParseException
{
    // JDK 7's try-with-resources
    try (FileReader reader = new FileReader("echo.json"))
    {
        JSONContext.Server jsonContext = (JSONContext.Server)bayeuxServer.getOption("jsonContext");
        EchoInfo info = jsonContext.getParser().parse(reader, EchoInfo.class);
        return info;
    }
}

public EchoInfo writeToConfigFile(BayeuxServer bayeuxServer, EchoInfo info) throws IOException
{
    // JDK 7's try-with-resources
    try (FileWriter writer = new FileWriter("echo.json"))
    {
        JSONContext.Server jsonContext = (JSONContext.Server)bayeuxServer.getOption("jsonContext");
        String json = jsonContext.getGenerator().generate(info);
        writer.write(json);
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
         version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <!-- other parameters -->
        <init-param>
            <param-name>jsonContext</param-name>
            <param-value>org.cometd.server.JacksonJSONContextServer</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

</web-app>

It is possible to switch from one implementation of the JSON library to another – for example from the Jetty library to the Jackson library, provided that you write the application code carefully.

As of version 1.8.4, Jackson can only produce instances of java.util.List when deserializing JSON arrays. The Jetty library, however, produces java.lang.Object[] when deserializing JSON arrays.

Similarly, Jackson produces java.lang.Integer where the Jetty library produces java.lang.Long.

To write portable application code, use the following code patterns:

Message message = ...;
Map<String, Object> data = message.getDataAsMap();

// Expecting a JSON array

// WRONG
Object[] array = (Object[])data.get("array");

// CORRECT
Object field = data.get("array");
Object[] array = field instanceof List ? ((List)field).toArray() : (Object[])field;


// Expecting a long

// WRONG
long value = (Long)data.get("value");

// CORRECT
long value = ((Number)data.get("value")).longValue();

Sometimes it is very useful to be able to obtain objects of application classes instead of just Map<String, Object> when calling message.getData().

You can easily achieve this with the Jetty JSON library. It is enough that the client formats the JSON object adding an additional class field whose value is the fully qualified class name that you want to convert the JSON to:

cometd.publish('/echo', {
    class: 'org.cometd.example.EchoInfo',
    id: '42',
    echo: 'cometd'
});

On the server, in the web.xml file, you register the org.cometd.server.JettyJSONContextServer as jsonContext (see Section 8.3.1.2, “Server Configuration”), and at startup you add a custom converter for the org.cometd.example.EchoInfo class (see Section 8.2.2.3, “Services Integration” for more details about configuring CometD at startup).

BayeuxServer bayeuxServer = ...;
JettyJSONContextServer jsonContext = (JettyJSONContextServer)bayeuxServer.getOption("jsonContext");
jsonContext.getJSON().addConvertor(EchoInfo.class, new EchoInfoConvertor());

Finally, these are the EchoInfoConvertor and EchoInfo classes:

public class EchoInfoConvertor implements JSON.Convertor
{
    public void toJSON(Object obj, JSON.Output out)
    {
        EchoInfo echoInfo = (EchoInfo)obj;
        out.addClass(EchoInfo.class);
        out.add("id", echoInfo.getId());
        out.add("echo", echoInfo.getEcho());
    }

    public Object fromJSON(Map map)
    {
        String id = (String)map.get("id");
        String echo = (String)map.get("echo");
        return new EchoInfo(id, echo);
    }
}

public class EchoInfo
{
    private final String id;
    private final String echo;

    public EchoInfo(String id, String echo)
    {
        this.id = id;
        this.echo = echo;
    }

    public String getId()
    {
        return id;
    }

    public String getEcho()
    {
        return echo;
    }
}

If, instead of using the JavaScript client, you are using the Java client, it is possible to configure the Java client to perform the serialization/deserialization of JSON objects in the same way (see also Section 8.3.1.1, “Client Configuration”):

JettyJSONContextClient jsonContext = ...;
jsonContext.getJSON().addConvertor(EchoInfo.class, new EchoInfoConvertor());

// Later in the application
BayeuxClient bayeuxClient = ...;

bayeuxClient.getChannel("/echo").subscribe(new ClientSessionChannel.MessageListener()
{
    public void onMessage(ClientSessionChannel channel, Message message)
    {
        // Receive directly EchoInfo objects
        EchoInfo data = (EchoInfo)message.getData();
    }
});

// Publish directly EchoInfo objects
bayeuxClient.getChannel("/echo").publish(new EchoInfo("42", "wohoo"));

A typical, but not the only, infrastructure to set up an Oort cluster is to have a load balancer in front of Oort nodes, so that clients can connect transparently to any node. The load balancer should implement stickyness, which can be based on:

You should configure DNS with a single host name/IP address pair (that of the load balancer), so that in case of a node crash, when clients attempt to reconnect to the same host name, the load balancer notices that the node has crashed and directs the connection to another node. The second node does not know about this client, and upon receiving the connect request sends to the client the advice to handshake.

The next sections use the following terminology:

Any CometD server can become an Oort comet by configuring an instance of org.cometd.oort.Oort. The org.cometd.oort.Oort instance is associated to the org.cometd.bayeux.server.BayeuxServer instance, and there can be only one Oort instance for each BayeuxServer instance.

Oort comets need to know each others' URLs to connect and form a cloud. A new comet that wants to join the cloud needs to know at least one URL of another comet that is already part of the cloud. Once it has connected to one comet, the cloud informs the new comet of the other comets to which the new comet is not yet connected, and the new comet then connects to all the existing nodes.

There are two ways for a new comet to discover at least one other comet:

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
     version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <servlet>
        <servlet-name>oort</servlet-name>
        <servlet-class>org.cometd.oort.OortMulticastConfigServlet</servlet-class>
        <load-on-startup>2</load-on-startup>
        <init-param>
            <param-name>oort.url</param-name>
            <param-value>http://host:port/context/cometd</param-value>
        </init-param>
    </servlet>
</web-app>

Destination    Gateway    Genmask      Flags    Metric    Ref    Use    Iface
224.0.0.0      0.0.0.0    240.0.0.0      U         0       0      0     eth0

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
     version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <servlet>
        <servlet-name>oort</servlet-name>
        <servlet-class>org.cometd.oort.OortStaticConfigServlet</servlet-class>
        <load-on-startup>2</load-on-startup>
        <init-param>
            <param-name>oort.url</param-name>
            <param-value>http://host:port/context/cometd</param-value>
        </init-param>
    </servlet>
</web-app>

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
     version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <servlet>
        <servlet-name>oort</servlet-name>
        <servlet-class>org.cometd.oort.OortStaticConfigServlet</servlet-class>
        <load-on-startup>2</load-on-startup>
        <init-param>
            <param-name>oort.url</param-name>
            <param-value>http://host1:port/context/cometd</param-value>
        </init-param>
        <init-param>
            <param-name>oort.cloud</param-name>
            <param-value>http://host2:port/context/cometd,http://host3:port/context/cometd</param-value>
        </init-param>
    </servlet>
</web-app>

public class OortConfigurationServlet extends GenericServlet
{
    public void init() throws ServletException
    {
        // Grab the Oort object
        Oort oort = (Oort)getServletContext().getAttribute(Oort.OORT_ATTRIBUTE);

        // Figure out the URLs to connect to, using other discovery means
        List<String> urls = ...;

        // Connect to the other Oort comets
        for (String url : urls)
        {
            OortComet oortComet = oort.observeComet(url);
            if (!oortComet.waitFor(1000, BayeuxClient.State.CONNECTED))
                throw new ServletException("Cannot connect to Oort comet " + url);
        }
    }

    public void service(ServletRequest request, ServletResponse response) throws ServletException, IOException
    {
        throw new ServletException();
    }
}

8.4.3.4. Membership Organization

When an Oort comet connects to another Oort comet, a bidirectional communication is established. If cometA connects to cometB (for example, via oortA.observeComet(urlB)), then an OortComet instance is created in cometA connected to cometB, and another OortComet instance is created in cometB connected to cometA.

After this direct, bidirectional communication has been established, a special message broadcasts across the whole Oort cloud (on channel /oort/cloud) where the two comets broadcast their known siblings. Every node receiving this message that does not know about those siblings then establishes a bidirectional communication with them.

For example, imagine that there are two simple Oort clouds, one made of comets A and B and the other made of comets C and D. When A and C connect, they broadcast their siblings (A broadcasts its siblings, now B and C, while C broadcasts its siblings, now A and D). All nodes connected, directly or indirectly, to the broadcaster receive this message. When C receives broadcasts from A's siblings it notices that one is itself (so it does nothing since it's already connected to A). The other is the unknown sibling B, and C establishes a bidirectional connection with B as well. Likewise, A receives the sibling broadcast message from C, and connects to D. Each new bidirectional connection triggers a sibling broadcast message on the whole cloud, until all comets are connected.

If a comet crashes, for example D, then all other comets detect that and disconnect from the faulty comet.

In this way, an Oort cloud is aware of its members, but it does not do anything useful for the application.

The next section covers broadcast message forwarding over the entire cloud.

Oort oort = ...;
oort.addCometListener(new Oort.CometListener()
{
    public void cometJoined(Event event)
    {
        System.out.printf("Comet joined the cloud %s%n", event.getCometURL());
    }

    public void cometLeft(Event event)
    {
        System.out.printf("Comet left the cloud %s%n", event.getCometURL());
    }
});

final Oort oort = ...;
oort.addCometListener(new Oort.CometListener()
{
    public void cometJoined(Event event)
    {
        String cometURL = event.getCometURL();
        OortComet oortComet = oort.observeComet(cometURL);

        // Push information to the new comet
        oortComet.getChannel("/service/foo").publish("bar");
    }

    public void cometLeft(Event event)
    {
    }
});

{
    "channel": "/meta/handshake",
    ... /* other usual handshake fields */
    "ext": {
        "org.cometd.oort": {
            "oortURL": "http://halley.cometd.org:8080/cometd",
            "cometURL": "http://halebopp.cometd.org:8080/cometd",
            "oortSecret": "cstw27r+l+XqE62IrNZdCDiUObA="
        }
    }
}

public class OortSecurityPolicy extends DefaultSecurityPolicy
{
    private final Oort oort;

    private OortSecurityPolicy(Oort oort)
    {
        this.oort = oort;
    }

    @Override
    public boolean canHandshake(BayeuxServer server, ServerSession session, ServerMessage message)
    {
        // Local sessions can always handshake
        if (session.isLocalSession())
            return true;

        // Remote Oort comets are allowed to handshake
        if (oort.isOortHandshake(message))
            return true;

        // Remote clients must have a valid token
        Map<String, Object> ext = message.getExt();
        return ext != null && isValid(ext.get("token"));
    }
}

8.4.3.7. Broadcast Messages Forwarding

Broadcast messages (that is, messages sent to non-meta and non-service channels, (see Section 7.3, “Subscribing and Unsubscribing” for further details) are by definition messages that all clients subscribing to the channel the message is being sent to should receive.

In an Oort cloud, you might have clients connected to different comets that subscribe to the same channel. If clientA connects to cometA, clientB connects to cometB and clientC connects to cometC, when clientA broadcasts a message you want clientB and clientC to receive, the Oort cloud must forward the message (sent by clientA and received by cometA) to cometB and cometC.

You accomplish this by configuring the Oort configuration servlets to set the oort.channels init parameter to a comma-separated list of channels whose messages are forwarded to the Oort cloud:

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
     version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <servlet>
        <servlet-name>oort</servlet-name>
        <servlet-class>org.cometd.oort.OortMulticastConfigServlet</servlet-class>
        <load-on-startup>2</load-on-startup>
        <init-param>
            <param-name>oort.url</param-name>
            <param-value>http://host1:port/context/cometd</param-value>
        </init-param>
        <init-param>
            <param-name>oort.channels</param-name>
            <param-value>/stock/**,/forex/*,/alerts</param-value>
        </init-param>
    </servlet>
</web-app>

Alternatively, you can use Oort.observeChannel(String channelName) to instruct a comet to listen for messages on that channel published to one of the known comets it is connected to.

When cometA observes a channel, it means that messages sent on that channel, but received by other comets, are automatically forwarded to cometA.

Important

Message forwarding is not bidirectional; if cometA forwards messages to other comets it is not automatic that the other comets forward messages to cometA. However, in most cases you configure the Oort comets in the same way by the same initialization code, and therefore all comets forward the same channels.

Forwarding of messages may be subject to temporary interruptions in case there is a temporary network connectivity failure between two comets. To overcome this problem, it is possible to configure the comets to enable the message acknowledgement extension (see Section 9.5, “Message Acknowledgment Extension”) so that, for short failures, the messages lost can be resent. Refer to Section 8.4.3.1, “Common Configuration” for the configuration details.

Remember that the message acknowledgement extension is not a fully persistent solution for lost messages (for example it does not guarantee message redelivery in case of long failures). CometD does not provide yet a fully persistent solution for messages in case of long failures.

Since it has the ability to observe messages published to broadcast channels, an Oort cloud can already implement a simple chat application among users connected to different nodes. In the example below, when clientA publishes a message on channel /chat (green arrow), it arrives on cometA; since cometB and cometC have been configured to observe channel /chat, they both receive the message from cometA (green arrows), and therefore they can deliver the chat message to clientB and clientC respectively (green arrows).

If your application only needs to broadcast messages to clients connected to other comets, an Oort instance is all you need.

If you need to send messages directly to particular clients (for example, clientA wants to send a message to clientC but not to clientB, then you need to set up an additional component of the Oort clustering called Seti, see Section 8.4.4, “Seti”.

Seti is the Oort clustering component that tracks clients connected to any comet in the cloud, and allows an application to send messages to particular client(s) in the cloud transparently, as if they were in the local comet.

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
     version="2.5">

    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.server.CometdServlet</servlet-class>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>cometd</servlet-name>
        <url-pattern>/cometd/*</url-pattern>
    </servlet-mapping>

    <servlet>
        <servlet-name>oort</servlet-name>
        <servlet-class>org.cometd.oort.OortServlet</servlet-class>
        <load-on-startup>2</load-on-startup>
        <init-param>
            <param-name>oort.url</param-name>
            <param-value>http://host:port/context/cometd</param-value>
        </init-param>
    </servlet>

    <servlet>
        <servlet-name>seti</servlet-name>
        <servlet-class>org.cometd.oort.SetiServlet</servlet-class>
        <load-on-startup>3</load-on-startup>
    </servlet>
</web-app>

public class MySecurityPolicy extends DefaultSecurityPolicy
{
    private final Seti seti;

    public MySecurityPolicy(Seti seti)
    {
        this.seti = seti;
    }

    @Override
    public boolean canHandshake(BayeuxServer server, ServerSession session, ServerMessage message)
    {
        if (session.isLocalSession())
            return true;

        // Authenticate
        String userId = performAuthentication(session, message);
        if (userId == null)
            return false;

        // Associate
        seti.associate(userId, session);

        return true;
    }
}

Seti seti = ...;
seti.addPresenceListener(new Seti.PresenceListener()
{
    public void presenceAdded(Event event)
    {
        System.out.printf("User ID %s is now present in comet %s%n", event.getUserId(), event.getURL());
    }

    public void presenceRemoved(Event event)
    {
        System.out.printf("User ID %s is now absent in comet %s%n", event.getUserId(), event.getURL());
    }
});

final Seti seti = ...;
seti.addPresenceListener(new Seti.PresenceListener()
{
    public void presenceAdded(Event event)
    {
        Oort oort = seti.getOort();
        String oortURL = event.getURL();
        OortComet oortComet = oort.getComet(oortURL);

        Map<String, Object> data = new HashMap<String, Object>
        data.put("action", "sync_request");
        data.put("userId", event.getUserId());

        oortComet.getChannel("/service/sync").publish(data);
    }

    public void presenceRemoved(Event event)
    {
    }
});

8.4.4.4. Sending Messages

After users have been associated, Seti.sendMessage(String userId, String channel, Object data) can send messages to a particular user in the cloud.

@Service("seti_forwarder");
public class SetiForwarder
{
    @Inject
    private Seti seti;

    @Listener("/service/forward")
    public void forward(ServerSession session, ServerMessage message)
    {
        Map<String,Object> data = message.getDataAsMap();
        String targetUserId = (String)data.get("targetUserId");
        seti.sendMessage(targetUserId, message.getChannel(), data);
    }
}

In the example below, clientA wants to send a message to clientC but not to clientB. Therefore clientA sends a message to the server it is connected to using a service channel so that the message is not broadcast (see Section 7.3, “Subscribing and Unsubscribing”), and then a specialized service (see Section 8.2.2, “Using Services”) routes the message to the appropriate user using Seti (see code snippet above). The Seti on cometA knows that the target user is on cometC (thanks to the association) and forwards the message to cometC, which in turn delivers the message to clientC.

Section 8.4, “Scalability Clustering with Oort” described how it is possible to link nodes to form an Oort cloud. Oort nodes are all connected to each other so that they are aware of all the other peer nodes. Additionally, each node can forward messages that have been published to broadcast channels (see Section 6.1.1, “Channel Definitions”) to other nodes.

Your application may need to maintain information, on each node, that is distributed across all nodes. A typical example is the total number of users connected to all nodes. Each node can easily know how many users are connected to itself, but in this case you want to know the total number of all users connected to all nodes (for example to display the number in a user interface). The information that you want to distribute is the "data entity" - in this case the number of users connected to each node - and we name this feature "data distribution". Having each node distributing its own data entity allows each node to know the data entity of the other nodes, and compute the total number of users.

Furthermore, your application may need to perform certain actions on a specific node. For example, your application may need to access a database system that is only accessible from a specific node for security reasons. We name this feature "service forwarding".

Oort and Seti (see Section 8.4.4, “Seti”) alone do not offer data distribution or service forwarding out of the box, but it is possible to build on Oort features to implement them, and this is exactly what CometD offers, respectively, with OortObject (Section 8.4.5.1, “OortObject”) and OortService (Section 8.4.5.2, “OortService”).

8.4.5.1. OortObject

An org.cometd.oort.OortObject instance represents a named composite data entity that is distributed in an Oort cluster. Each node in the cluster has exactly one instance of an OortObject with a specific name, and the OortObject contains N data entities, one of each node in the cluster. There may be several Oort objects in the same node, provided they all have different names.

The data entity may be the number of users connected to each node, or the number of games played on each node, or the list of chat rooms created on each node, or the names of systems monitored by each node, etc., depending on your application's business domain.

In the image below, you can see 3 nodes (nodeA, nodeB and nodeC), each containing an Oort object named "users" (in orange) that stores the names of the users connected to each Oort node. The data entity in this case is a List<String> representing the names of the connected users for each node.

NodeA has clients Ca1 and Ca2 connected, nodeB has only client Cb1 connected, while nodeC has 3 clients connected. Oort objects are composites in that they store N data entities, also called parts, where N is the number of nodes in the Oort cloud. You can see that each Oort object is made of 3 parts (the innermost blue, green and red boxes); each part is colored like the node it represents. The part that has the same color as the node it lives in it's the local part.

Each Oort object can only update its local part: nodeA can only add/remove user names from its local (blue) part, and cannot add/remove from the remote parts (in green and red). Likewise, nodeB can only update the green part but not the blue and red parts, and nodeC can only update the red part, but not the blue and green ones.

If a new client connects to nodeB, say Cb2, then the application on nodeB takes the user name (B2) that wants to share with other nodes, and adds it to the Oort object on nodeB. The user name B2 will be added to the green part of nodeB, and a message will be broadcast to the other nodes, which will also modify the correspondent green parts on themselves, adding a copy of B2. The remote parts of an Oort object can only be updated by messages internal to the OortObject implementation; they cannot be updated by application code directly.

Each Oort object instance owns only its local part. In the example, the user name A2 is present in all the nodes, but it is owned only by the Oort object in nodeA. Anyone that wants to modify or remove A2 must perform this action in nodeA. Section 8.4.5.2, “OortService” shows how to forward service actions from one node to another.

OortObject allows applications to add/remove OortObject.Listeners that are notified of modification of a part, either local or remote. Your application can implement these listeners to perform custom logic, see Section 8.4.5.1.6, “OortObject Listeners”.

8.4.5.1.1. OortObject Specializations

While OortObject is a generic container of objects (like a List<String>), it may not be very efficient. Imagine the case where the list contains thousands of names: the addition/removal of one name will cause the whole list to be replicated to all other nodes, because the whole list is the data entity.

To avoid this inefficiency, CometD offers these specializations of OortObject:

• OortMap, an OortObject that contains a ConcurrentMap

• OortStringMap, an OortMap with String keys

• OortLongMap, an OortMap with Long keys

• OortList, an OortObject that contains a List

Each specialization replicates single operations like the addition/removal of a key/value pair in an OortMap, or the addition/removal of an element in an OortList.

OortMap provides an OortMap.EntryListener to notify applications of map entry addition/removal, either local or remote. OortList provides an OortList.ElementListener to notify applications of element addition/removal, either local or remote. Applications can implement these listeners to be notified of entry or element updates in order to perform custom logic, see Section 8.4.5.1.6, “OortObject Listeners”.

8.4.5.1.2. OortObject Creation

OortObjects are created by providing an OortObject.Factory. This factory is needed to create the data entity from its raw representation obtained from JSON. This allows standard JDK containers such as java.util.concurrent.ConcurrentHashMap to be used as data entities, but replicated among nodes using standard JSON.

CometD provides a number of predefined factories in class org.cometd.oort.OortObjectFactories, for example:

Oort oort = ...;

// The factory for data entities
OortObject.Factory<List<String>> factory = OortObjectFactories.forList();

// Create the OortObject
OortObject<List<String>> users = new OortObject<List<String>>(oort, "users", factory);

// Start it before using it
users.start();

The code above will create an OortObject named "users" whose data entity is a List<String> (this is an example; you may want to use the richer and more powerful OortList<String> instead). Once you have created an OortObject you must start it before using it.

OortObjects are usually created at startup time, so that all the nodes have one instance of the same OortObjects with the same names. Remember that the data entity is distributed among OortObjects with the same name, so if a node does not have that particular named OortObject, then it will not receive updates for that data entity.

It is possible to create OortObjects on-the-fly in response to some event, but the application must make sure that this event is broadcast to all nodes so that each node can create its own OortObject instance with the same name.

8.4.5.1.3. OortObject Data Entity Sharing

One OortObject owns one data entity, which is its local part. In the example above, the data entity is a whole List<String>, so that's what we want to share with other nodes:

OortObject.Factory<List<String>> factory = users.getFactory();

// Create a "default" data entity
List<String> names = factory.newObject(null);

// Fill it with data
names.add("B1");

// Share the new list with the other nodes
users.setAndShare(names);

Method setAndShare(...) will replace the empty list (created internally when the OortObject was created) with the provided list, and broadcast this event to the cluster so that other nodes can replace the part they have associated with this node with the new one.

Similarly, OortMap has the putAndShare(...) and removeAndShare(...) methods to put/remove the map entries and share them:

OortStringMap<UserInfo> userInfos = ...;

// Map user name "B1" with its metadata
userInfos.putAndShare("B1", new UserInfo("B1", ...));

// In another place in the code

// Remove the mapping for user "B1"
userInfos.removeAndShare("B1");

OortList has addAndShare(...) and removeAndShare(...):

OortList<String> names = ...;

// Add user name "B1"
names.addAndShare("B1");

// In another place in the code

// Remove user "B1"
names.removeAndShare("B1");

Both OortMap and OortList inherit from OortObject method setAndShare(...) if you need to replace the whole map or list.

The OortObject API will try to make it hard for you to interact directly with the data entity, and this is by design. If you can modify the data entity directly without using the above methods, then the local data entity will be out of sync with the correspondent data entities in the other nodes. Whenever you feel the need to access the data entity, and you cannot find an easy way to do it, consider that you are probably taking the wrong approach.

For the same reasons mentioned above, it is highly recommended that the data that you store in an Oort object is immutable. In the OortStringMap example above, the UserInfo object should be immutable, and if you need to change it, it is better to create a new UserInfo instance with the new data and then call putAndShare(...) to replace the old one, which will ensure that all nodes will get the update.

8.4.5.1.4. OortObject Custom Data Entity Serialization

The OortObject implementation must be able to transmit and receive the data entity to/from other nodes in the cluster, and recursively so for all objects contained in the data entity that is being transmitted.

The data entity and the objects it contains are serialized to JSON using the standard CometD mechanism, and then transmitted. When a node receives the JSON representation of data entity and its contained objects, it deserializes it from JSON into an object graph.

In the OortStringMap example above, the data entity is a ConcurrentMap<String, Object> and the values of this data entity are objects of class UserInfo.

While the OortObject implementation is able to serialize a ConcurrentMap to JSON natively (because ConcurrentMap is a Map and therefore has a native representation as a JSON object), it usually cannot serialize UserInfo instances correctly (by default, CometD just calls toString() to convert such non natively representable objects to JSON).

In order to serialize correctly instances of UserInfo, you must configure Oort as explained in Section 8.3.1.3, “Oort Configuration”. This is done by creating a custom implementation of JSONContent.Client:

package com.acme;

import org.cometd.common.JettyJSONContextClient;

public class MyCustomJSONContextClient extends JettyJSONContextClient
{
    public MyCustomJSONContextClient()
    {
        getJSON().addConvertor(UserInfo.class, new UserInfoConvertor());
    }
}

In the example above the Jetty JSON library has been implicitly chosen by extending the CometD class JettyJSONContextClient. A similar class exist for the Jackson JSON library. In the class above a convertor for the UserInfo class is added to the root org.eclipse.jetty.util.ajax.JSON object retrieved via getJSON(). This root JSON object is the one responsible for CometD message serialization.

A typical implementation of the convertor could be (assuming that your UserInfo class has an id property):

import java.util.Map;
import org.eclipse.jetty.util.ajax.JSON;

public class UserInfoConvertor implements JSON.Convertor
{
    @Override
    public void toJSON(Object obj, JSON.Output out)
    {
        UserInfo userInfo = (UserInfo)obj;
        out.addClass(UserInfo.class);
        out.add("id", userInfo.getId());
    }

    @Override
    public Object fromJSON(Map object)
    {
        String id = (String)object.get("id");
        return new UserInfo(id);
    }
}

Class UserInfoConvertor depends on the Jetty JSON library; a similar class can be written for the Jackson library (refer to Section 8.3, “JSON Libraries” for further information).

Finally, you must specify class MyCustomJSONContextClient as the jsonContext parameter of the Oort configuration (as explained in Section 8.4.3.1, “Common Configuration”) in the web.xml file, for example:

<web-app ... >
    ...
    <servlet>
        <servlet-name>oort-config</servlet-name>
        <servlet-class>org.cometd.oort.OortMulticastConfigServlet</servlet-class>
        <init-param>
            <param-name>oort.url</param-name>
            <param-value>http://localhost:8080/cometd</param-value>
        </init-param>
        <init-param>
            <param-name>oort.secret</param-name>
            <param-value>oort_secret</param-value>
        </init-param>
        <init-param>
            <param-name>jsonContext</param-name>
            <param-value>com.acme.MyCustomJSONContextClient</param-value>
        </init-param>
        <load-on-startup>2</load-on-startup>
    </servlet>
    ...
</web-app>

Similarly, in order to deserialize correctly instances of UserInfo, you must configure CometD, again as explained in Section 8.3.1.3, “Oort Configuration”. This is done by creating a custom implementation of JSONContext.Server:

package com.acme;

import org.cometd.server.JettyJSONContextServer;

public class MyCustomJSONContextServer extends JettyJSONContextServer
{
    public MyCustomJSONContextServer()
    {
        getJSON().addConvertor(UserInfo.class, new UserInfoConvertor());
    }
}

Like before, the Jetty JSON library has been implicitly chosen by extending the CometD class JettyJSONContextServer. A similar class exist for the Jackson JSON library. Class UserInfoConvertor is the same class we defined above and it is therefore used for both serialization and deserialization.

You must specify class MyCustomJSONContextServer as the jsonContext parameter of the CometD configuration (as explained in Section 8.2.1, “Configuring the Java Server”) in the web.xml file, for example:

<web-app ... >
    ...
    <servlet>
        <servlet-name>cometd</servlet-name>
        <servlet-class>org.cometd.annotation.AnnotationCometdServlet</servlet-class>
        <init-param>
            <param-name>jsonContext</param-name>
            <param-value>com.acme.MyCustomJSONContextServer</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>
    ...
</web-app>

To summarize, the serialization of the ConcurrentMap data entity of a OortStringMap will happen in the following way: the ConcurrentMap is a Map and is natively represented as a JSON object; the UserInfo values will be converted to JSON as specified by the UserInfoConvertor.toJSON(...) method.

The JSON obtained after the serialization is transmitted to other nodes. The node that receive it will deserialize the received JSON into a plain Map containing UserInfo value objects converted as specified by the UserInfoConvertor.fromJSON(...) method. Finally the plain Map object will be passed to the Oort object factory (see Section 8.4.5.1.2, “OortObject Creation”) to be converted into a ConcurrentMap.

8.4.5.1.5. OortObject Data Entity Merging

OortObjects are made of parts, and applications may need to access the data contained in all parts. In the examples above, an application may want to be able to access all the user names from all nodes.

In order to access the data from all parts, OortObject provides the merge(OortObject.Merger merger) method. Applications can use mergers provided by org.cometd.oort.OortObjectMergers or implement their own, for example:

OortList<String> names = ...;

// Merge all the names from all the nodes
List<String> allNames = names.merge(OortObjectMergers.listUnion());

Merging is a local operation that does not involve network communication: it is just merging all the data entity parts contained in the OortObject.

8.4.5.1.6. OortObject Listeners

When one node updates the data entity it owns, CometD notifies the other nodes so that they can keep in sync the data entity part correspondent to the node that performed the update. Applications can register listeners to be notified of such events, and perform their custom logic.

A typical example is when an application needs to show the total number of currently logged in users. Every time a user connects and logs in, say, in NodeA, then NodeB needs to be notified to update the total number in the user interface of the users connected to NodeB. The Oort object we use in this example is an OortObject<Long>, but you want to use CometD's built-in org.cometd.oort.OortLong in your application.

Since the application already updates the OortObject<Long> in NodeA, the correspondent OortObject<Long> in NodeB is updated too. The application can register a listener for such events, and update the user interface:

// At initialization time, create the OortObject and add the listener
final OortObject<Long> userCount = new ...;
userCount.addListener(new OortObject.Listener()
{
    public void onUpdated(OortObject.Info<T> oldInfo, OortObject.Info<T> newInfo)
    {
        // The user count changed somewhere, broadcast the new value
        long count = userCount.merge(OortObjectMergers.longSum());
        broadcastUserCount(count);
    }

    public void onRemoved(OortObject.Info<T> info);
    {
        // A node disappeared, broadcast the new user count
        long count = userCount.merge(OortObjectMergers.longSum());
        broadcastUserCount(count);
    }

    private void broadcastUserCount(long count)
    {
        // Publish a message on "/user/count" to update the remote clients connected to this node
        BayeuxServer bayeuxServer = userCount.getOort().getBayeuxServer();
        bayeuxServer.getChannel("/user/count").publish(userCount.getLocalSession(), count, null);
    }
});

Class org.cometd.oort.OortObject.Info represents a data entity part of an OortObject and contains the data entity and the Oort URL correspondent to the node that it represent. For this particular example, the Info objects are not important, since we are only interested in the total user count, that can be obtained by merging (see Section 8.4.5.1.5, “OortObject Data Entity Merging”). They can be used, however, to compute the difference before and after the update if needed.

Similarly, OortMap supports registration of OortMap.EntryListeners that are notified when OortMap entries change due to calls to putAndShare(...) or removeAndShare(...). OortMap.EntryListener are notified only when map entries are updated. To be notified when the whole map changes due to calls to setAndShare(...), you can use an OortMap.Listener (inherited from OortObject) as described above. In some cases, the whole map is updated but you want to be notified as if single entries are changed; in this case you can use an OortMap.DeltaListener, that converts whole map updates into map entry updates.

OortList supports registration of OortList.ElementListeners that are notified when OortList elements change due to calls to addAndShare(...) or removeAndShare(...). OortList.ElementListener are notified only when list elements are updated. To be notified when the whole list changes due to calls to setAndShare(...), you can use an OortList.Listener (inherited from OortObject) as described above. In some cases, the whole list is updated but you want to be notified as if single elements are changed; in this case you can use an OortList.DeltaListener, that converts whole list updates into list element updates.

ServerAnnotationProcessor processor = ...;
Oort oort = ...;

// Create the service instance and process its annotations
NameEditService nameEditService = new NameEditService(oort);
processor.process(nameEditService);

@Service(NameEditService.NAME)
public class NameEditService extends OortService<String, Boolean>
{
    public static final String NAME = "name_edit";

    public NameEditService(Oort oort)
    {
        super(oort, NAME);
    }

    // Lifecycle methods triggered by standard lifecycle annotations

    @PostConstruct
    public void construct() throws Exception
    {
        start();
    }

    @PreDestroy
    public void destroy() throws Exception
    {
        stop();
    }

    // CometD's listener method on channel "/service/edit"

    @org.cometd.annotation.Listener("/service/edit")
    public void editName(final ServerSession remote, final ServerMessage message)
    {
        // Step #1: remote client sends an action request.
        // This runs in the requesting node.

        // Find the owner Oort URL from the message.
        // Applications must implement this method with their logic.
        String ownerURL = findOwnerURL(message);

        // Prepare the action data.
        String oldName = (String)remote.getAttribute("name");
        String newName = (String)message.getDataAsMap().get("name");
        Map<String, Object> actionData = new HashMap<String, Object>();
        actionData.put("oldName", oldName);
        actionData.put("newName", newName);

        // Step #2: forward to the owner node.
        // Method forward(...) is inherited from OortService
        forward(ownerURL, actionData, new ServerContext(remote, message));
    }

    @Override
    protected Result<Boolean> onForward(Request request)
    {
        // Step #3: perform the action.
        // This runs in the owner node.

        try
        {
            Map<String, Object> actionData = request.getDataAsMap();
            // Edit the name.
            // Applications must implement this method.
            boolean result = editName(actionData);

            // Return the action result.
            return Result.success(result);
        }
        catch (Exception x)
        {
            // Return the action failure.
            return Result.failure("Could not edit name, reason: " + x.getMessage());
        }
    }

    @Override
    protected void onForwardSucceeded(Boolean result, ServerContext context)
    {
        // Step #4: successful result.
        // This runs in the requesting node.

        // Notify the remote client of the result of the edit.
        context.getServerSession().deliver(getLocalSession(), result, null);
    }

    @Override
    protected void onForwardFailed(Object failure, ServerContext context)
    {
        // Step #5: failure result.
        // This runs in the requesting node.

        // Notify the remote client of the failure.
        context.getServerSession().deliver(getLocalSession(), failure, null);
    }
}

To enable the ActivityExtension, you must add the extension to the BayeuxServer during initialization, specifying the type of activity you want to monitor, and the max inactivity period in milliseconds:

bayeuxServer.addExtension(new ActivityExtension(ActivityExtension.Activity.CLIENT, 15 * 60 * 1000L));

The example above configures the ActivityExtension to monitor client-only activity, and disconnects inactive clients after 15 minutes (or 15 * 60 * 1000 ms) of inactivity.

Similarly, to monitor client-server activity:

bayeuxServer.addExtension(new ActivityExtension(ActivityExtension.Activity.CLIENT_SERVER, 15 * 60 * 1000L));

The example above configures the ActivityExtension to monitor client-server activity, and disconnects inactive clients after 15 minutes of inactivity.

The org.cometd.server.ext.ActivityExtension can be installed if you want to monitor the inactivity of all clients in the same way.

It is possible to monitor the inactivity of particular clients by not installing the ActivityExtension on the BayeuxServer, and by installing a org.cometd.server.ext.ActivityExtension.SessionExtension on the specific server session for which you want to monitor inactivity, for example:

public class MyPolicy extends DefaultSecurityPolicy
{
    @Override
    public boolean canHandshake(BayeuxServer server, ServerSession session, ServerMessage message)
    {
        if (!isAdminUser(session, message))
            session.addExtension(new ActivityExtension.SessionExtension(ActivityExtension.Activity.CLIENT, 10 * 60 * 1000L));
        return true;
    }
}

In the example above, a custom SecurityPolicy (see Section 8.2.3, “Authorization”) checks whether the handshake is that of an administrator user, and installs the activity extension only for non-administrators, with an inactivity timeout of 10 minutes.

Alternatively, you can write a BayeuxServer extension that installs the ActivityExtension.SessionExtension selectively:

public class MyExtension extends BayeuxServer.Extension.Adapter
{
    @Override
    public boolean sendMeta(ServerSession session, ServerMessage.Mutable message)
    {
        if (Channel.META_HANDSHAKE.equals(message.getChannel()) && message.isSuccessful())
        {
            if (!isAdminUser(session, message))
                session.addExtension(new ActivityExtension.SessionExtension(ActivityExtension.Activity.CLIENT, 10 * 60 * 1000L));
        }
        return true;
    }
}

In the example above, a custom BayeuxServer extension checks whether the handshake has been successful, and installs the activity extension only for non-administrators, with an inactivity timeout of 10 minutes.

To enable support for acknowledged messages, you must add the extension to the org.cometd.bayeux.server.BayeuxServer instance during initialization:

bayeuxServer.addExtension(new org.cometd.server.ext.AcknowledgedMessagesExtension());

The AcknowledgedMessageExtension is a per-server extension that monitors handshakes from new remote clients, looking for those that also support the acknowledged message extension, and then adds the AcknowledgedMessagesClientExtension to the ServerSession correspondent to the remote client, during the handshake processing.

Once added to a ServerSession, the AcknowledgedMessagesClientExtension guarantees ordered delivery of messages, and resend of unacknowledged messages, from server to client. The extension also maintains a list of unacknowledged messages and intercepts the traffic on the /meta/connect channel to insert and check acknowledge IDs.

The dojox/cometd/ack.js provides the client-side extension binding for Dojo, and it is sufficient to use Dojo's dojo.require mechanism:

dojo.require("dojox.cometd.ack");

The file jquery.cometd-ack.js provides the client-side extension binding for jQuery. You must include this file in the HTML page via the <script> tag:

<script type="text/javascript" src="AckExtension.js"></script>
<script type="text/javascript" src="jquery.cometd-ack.js"></script>

In both Dojo and jQuery extension bindings, the extension is registered on the default cometd object under the name ack.

Furthermore, you can programmatically disable/enable the extension before initialization by setting the ackEnabled boolean field on the cometd object:

// Disables the ack extension during handshake
cometd.ackEnabled = false;
cometd.init(cometdURL);

To enable message acknowledgement, both client and server must indicate that they support message acknowledgement. This is negotiated during handshake. On handshake, the client sends {"ext":{"ack": "true"}} to indicate that it supports message acknowledgement. If the server also supports message acknowledgement, it likewise replies with {"ext":{"ack": "true"}}.

The extension does not insert ack IDs in every message, as this would impose a significant burden on the server for messages sent to multiple clients (which would need to be reserialized to JSON for each client). Instead the ack ID is inserted in the ext field of the /meta/connect messages that are associated with message delivery. Each /meta/connect request contains the ack ID of the last received ack response: "ext":{"ack": 42}. Similarly, each /meta/connect response contains an ext ack ID that uniquely identifies the batch of responses sent.

If a /meta/connect message is received with an ack ID lower that any unacknowledged messages held by the extension, then these messages are requeued prior to any more recently queued messages and the /meta/connect response sent with a new ack ID.

It is important to note that message acknowledgement is guaranteed from server to client only, and not from client to server. This means that the ack extension guarantees that messages sent by the server to the clients will be resent in case of temporary network failures that produce loss of messages. It does not guarantee however, that messages sent by the client will reach the server.

Message ordering is not enforced by default by CometD. A CometD server has two ways to deliver the messages present in the ServerSession queue to its correspondent remote client:

Delivery through a /meta/connect response means that the server will deliver the messages present in the ServerSession queue along with a /meta/connect response, so that the messages delivered to the remote client are: [{/meta/connect response message}, {queued message 1}, {queued message 2}, ...].

Direct delivery depends on the transport.

For polling transports it means that the server will deliver the messages present in the ServerSession queue along with some other response message that is being processed in that moment. For example, let's assume that clientA is already subscribed to channel /foo, and that it wants to subscribe also to channel /bar. Then clientA sends a subscription request for channel /bar to the server, and just before processing the subscription request for channel/bar, the server receives a message on channel /foo, that therefore needs to be delivered to clientA (for example, clientB published a message to channel/foo, or an external system produced a message on channel /foo). The message on channel /foo gets queued on clientA's ServerSession. However, the server notices that it has to reply to subscription request for /bar, so it includes the message on channel /foo in that response, thus avoiding to wake up the /meta/connect, so that the messages delivered to the remote client are: [{subscription response message for /bar}, {queued message on /foo}].

For non-polling transports such as websocket the server will just deliver the messages present in the ServerSession queue without waking up the /meta/connect, because non-polling transports have a way to send server-to-client messages without the need to have a pending response onto which the ServerSession's queued messages are piggybacked.

These two ways of delivering messages compete each other to deliver messages with the smallest latency. Therefore it is possible that a server receives from an external system two messages to be delivered for the same client, say message1 first and then message2; message1 is queued and immediately dequeued by a direct delivery, while message2 is queued and then dequeued by a /meta/connect delivery. The client may see message2 arriving before message1, for example because the thread scheduling on the server favored message2's processing or because the TCP communication for message1 was somehow slowed down (not to mention that browsers could as well be source of uncertainty).

To enable just server-to-client message ordering (but not reliability), you need to configure the server with the metaConnectDeliverOnly parameter, as explained in Section 8.2.1, “Configuring the Java Server”.

When server-to-client message ordering is enabled, all messages will be delivered through meta/connect responses. In the example above, message1 will be delivered to the client, and message2 will wait on the server until another meta/connect is issued by the client (which happens immediately after message1 has been received by the client). When the server receives the second meta/connect request, it will respond to it immediately and deliver message2 to the client.

It is clear that server-to-client message ordering comes at the small price of slightly increased latencies (message2 has to wait the next meta/connect before being delivered), and slightly more activity for the server (since meta/connects will be resumed more frequently than normal).

There is an example of acknowledged messages in the Dojo chat demo that comes bundled with the CometD distribution, and instruction on how to run the CometD demos in Section 3.1, “Running the Demos”.

To run the acknowledgement demo, follow these steps:

$ cd cometd-demo
$ mvn jetty:run

Notice that if the disconnected browser is disconnected in excess of maxInterval (default 10s), the client times out and the unacknowledged queue is discarded.

The dojox/cometd/reload.js provides the client side extension binding for Dojo, and it is sufficient to use Dojo's dojo.require mechanism:

dojo.require("dojox.cometd.reload");

The file jquery.cometd-reload.js provides the client-side extension binding for jQuery. This file must be included in the HTML page via the <script> tag, along with the jQuery cookie plugin and the reload extension implementation:

<script type="text/javascript" src="jquery.cookie.js"></script>
<script type="text/javascript" src="ReloadExtension.js"></script>
<script type="text/javascript" src="jquery.cometd-reload.js"></script>