How to make a Pub/Sub service with CometD and Jetty

Question

I need to create a Java 8- based service that provides a CometD channel that multiple clients can subscribe to. The idea is that the server can send notifications to the clients when certain events occur.

I am using Jetty 9 as my servlet container (necessary to meet the requirements for my group). I have been reading CometD documentation and looking for some kind of example that I can use. The documentation is extensive but isn't helping (lack of context), and I haven't been able to find a decent example of what I am trying to do.

Can someone provide a simple example of creating a publication mechanism, in Java, that can be used with Jetty? Failing that, can someone point me to an example of how to do it?

Please advise.

Answer 1

The CometD Project has an outstanding task to bring back the tutorials.

This particular question was answered by the server-side stock price tutorial, for which you can find the source here while we work on it to bring it back online as part of the documentation.

Glossing over a few details, the service you need to write is similar to the tutorial's stock price service: upon receiving an external event, the service should broadcast the event to subscribers.

@Service
public class StockPriceService implements StockPriceEmitter.Listener
{
    @Inject
    private BayeuxServer bayeuxServer;
    @Session
    private LocalSession sender;

    public void onUpdates(List<StockPriceEmitter.Update> updates)
    {
        for (StockPriceEmitter.Update update : updates)
        {
            // Create the channel name using the stock symbol.
            String channelName = "/stock/" + update.getSymbol().toLowerCase(Locale.ENGLISH);

            // Initialize the channel, making it persistent and lazy.
            bayeuxServer.createChannelIfAbsent(channelName, new ConfigurableServerChannel.Initializer()
            {
                public void configureChannel(ConfigurableServerChannel channel)
                {
                    channel.setPersistent(true);
                    channel.setLazy(true);
                }
            });

            // Convert the Update business object to a CometD-friendly format.
            Map<String, Object> data = new HashMap<>(4);
            data.put("symbol", update.getSymbol());
            data.put("oldValue", update.getOldValue());
            data.put("newValue", update.getNewValue());

            // Publish to all subscribers.
            ServerChannel channel = bayeuxServer.getChannel(channelName);
            channel.publish(sender, data);
        }
    }
}

Class StockPriceEmitter is the source of your external events, and publishes them to StockPriceEmitter.Listener in form of StockPriceEmitter.Update events.

How the external events are relayed to the CometD server is the detail that StockPriceEmitter hides; it could be done via JMS messages, or by polling an external REST service, or via a custom network protocol, or by polling a database, etc.

The important thing is that when the external events arrive, StockPriceService.onUpdates(...) is called, and there you can convert the events into a CometD friendly JSON format, and then publish them to the CometD channel.

Publishing to the CometD channel, in turn, will send the message to all subscribers for that channel, typically remote clients such as browsers.

The CometD channel has been made lazy because it is a way to avoid bombing the clients with a very frequent update rate (say, higher than 2-4 updates per second). You will need to decide about the lazyness of the channel based on your particular use case.