Chapter 14. Server-Sent Events (SSE) Support

Firstly you need to add a Jersey SSE module dependency to your project as shown in the earlier section and register the SseFeature from this module in your Application or ResourceConfig. Once done, you are ready to add SSE support to your resource:

...
import org.glassfish.jersey.media.sse.EventOutput;
import org.glassfish.jersey.media.sse.OutboundEvent;
import org.glassfish.jersey.media.sse.SseFeature;
...

@Path("events")
public static class SseResource {

    @GET
    @Produces(SseFeature.SERVER_SENT_EVENTS)
    public EventOutput getServerSentEvents() {
        final EventOutput eventOutput = new EventOutput();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    for (int i = 0; i < 10; i++) {
                        // ... code that waits 1 second
                        final OutboundEvent.Builder eventBuilder
                        = new OutboundEvent.Builder();
                        eventBuilder.name("message-to-client");
                        eventBuilder.data(String.class,
                            "Hello world " + i + "!");
                        final OutboundEvent event = eventBuilder.build();
                        eventOutput.write(event);
                    }
                } catch (IOException e) {
                    throw new RuntimeException(
                        "Error when writing the event.", e);
                } finally {
                    try {
                        eventOutput.close();
                    } catch (IOException ioClose) {
                        throw new RuntimeException(
                            "Error when closing the event output.", ioClose);
                    }
                }
            }
        }).start();
        return eventOutput;
    }
}

The code above defines the resource deployed on URI "/events". This resource has a single @GET resource method which returns as an entity EventOutput - an extension of generic Jersey ChunkedOutput API for output chunked message processing.

After the eventOutput is returned from the method, the Jersey runtime recognizes that this is a ChunkedOutput extension and does not close the client connection immediately. Instead, it writes the HTTP headers to the response stream and waits for more chunks (SSE events) to be sent. At this point the client can read headers and starts listening for individual events.

In the Example 14.2, “Simple SSE resource method”, the resource method creates a new thread that sends a sequence of 10 events. There is a 1 second delay between two subsequent events as indicated in a comment. Each event is represented by OutboundEvent type and is built with a helpf of an outbound event Builder. The OutboundEvent reflects the standardized format of SSE messages and contains properties that represent name (for named events), comment, data or id. The code also sets the event data media type using the mediaType(MediaType) method on the eventBuilder. The media type, together with the data type set by the data(Class, Object> method (in our case String.class), is used for serialization of the event data. Note that the event data media type will not be written to any headers as the response Content-type header is already defined by the @Produces and set to "text/event-stream" using constant from the SseFeature. The event media type is used for serialization of event data. Event data media type and Java type are used to select the proper MessageBodyWriter<T> for event data serialization and are passed to the selected writer that serializes the event data content. In our case the string "Hello world " + i + "!" is serialized as "text/plain". In event data you can send any Java entity and associate it with any media type that you would be able to serialize with an available MessageBodyWriter<T>. Typically, you may want to send e.g. JSON data, so you would fill the data with a JAXB annotated bean instance and define media type to JSON.

Once an outbound event is ready, it can be written to the eventOutput. At that point the event is serialized by internal OutboundEventWriter which uses an appropriate MessageBodyWriter<T> to serialize the "Hello world " + i + "!" string. You can send as many messages as you like. At the end of the thread execution the response is closed which also closes the connection to the client. After that, no more messages can be sent to the client on this connection. If the client would like to receive more messages, it would have to send a new request to the server to initiate a new SSE streaming connection.

A client connecting to our SSE-enabled resource will receive the following data from the entity stream:

event: message-to-client
data: Hello world 0!

event: message-to-client
data: Hello world 1!

event: message-to-client
data: Hello world 2!

event: message-to-client
data: Hello world 3!

event: message-to-client
data: Hello world 4!

event: message-to-client
data: Hello world 5!

event: message-to-client
data: Hello world 6!

event: message-to-client
data: Hello world 7!

event: message-to-client
data: Hello world 8!

event: message-to-client
data: Hello world 9!

Each message is received with a delay of one second.

Jersey SSE server API defines SseBroadcaster which allows to broadcast individual events to multiple clients. A simple broadcasting implementation is shown in the following example:

...
import org.glassfish.jersey.media.sse.SseBroadcaster;
...

@Singleton
@Path("broadcast")
public static class BroadcasterResource {

    private SseBroadcaster broadcaster = new SseBroadcaster();

    @POST
    @Produces(MediaType.TEXT_PLAIN)
    @Consumes(MediaType.TEXT_PLAIN)
    public String broadcastMessage(String message) {
        OutboundEvent.Builder eventBuilder = new OutboundEvent.Builder();
        OutboundEvent event = eventBuilder.name("message")
            .mediaType(MediaType.TEXT_PLAIN_TYPE)
            .data(String.class, message)
            .build();

        broadcaster.broadcast(event);

        return "Message '" + message + "' has been broadcast.";
    }

    @GET
    @Produces(SseFeature.SERVER_SENT_EVENTS)
    public EventOutput listenToBroadcast() {
        final EventOutput eventOutput = new EventOutput();
        this.broadcaster.add(eventOutput);
        return eventOutput;
    }
}

Let's explore the example together. The BroadcasterResource resource class is annotated with @Singleton annotation which tells Jersey runtime that only a single instance of the resource class should be used to serve all the incoming requests to /broadcast path. This is needed as we want to keep an application-wide single reference to the private broadcaster field so that we can use the same instance for all requests. Clients that want to listen to SSE events first send a GET request to the BroadcasterResource, that is handled by the listenToBroadcast() resource method. The method creates a new EventOutput representing the connection to the requesting client and registers this eventOutput instance with the singleton broadcaster, using its add(EventOutput) method. The method then returns the eventOutput which causes Jersey to bind the eventOutput instance with the requesting client and send the response HTTP headers to the client. The client connection remains open and the client is now waiting ready to receive new SSE events. All the events are written to the eventOutput by broadcaster later on. This way developers can conveniently handle sending new events to all the clients that subscribe to them.

When a client wants to broadcast new message to all the clients listening on their SSE connections, it sends a POST request to BroadcasterResource resource with the message content. The method broadcastMessage(String) is invoked on BroadcasterResource resource with the message content as an input parameter. A new SSE outbound event is built in the standard way and passed to the broadcaster. The broadcaster internally invokes write(OutboundEvent) on all registered EventOutputs. After that the method just return a standard text response to the POSTing client to inform the client that the message was successfully broadcast. As you can see, the broadcastMessage(String) resource method is just a simple JAX-RS resource method.

In order to implement such a scenario, you may have noticed, that the Jersey SseBroadcaster is not mandatory to complete the use case. individual EventOutputs can be just stored in a collection and iterated over in the broadcastMessage method. However, the SseBroadcaster internally identifies and handles also client disconnects. When a client closes the connection the broadcaster detects this and removes the stale connection from the internal collection of the registered EventOutputs as well as it frees all the server-side resources associated with the stale connection. Additionally, the SseBroadcaster is implemented to be thread-safe, so that clients can connect and disconnect in any time and SseBroadcaster will always broadcast messages to the most recent collection of registered and active set of clients.

The events can be read on the client side from a EventInput. See the following code:

Client client = ClientBuilder.newBuilder()
        .register(SseFeature.class).build();
WebTarget target = client.target("http://localhost:9998/events");

EventInput eventInput = target.request().get(EventInput.class);
while (!eventInput.isClosed()) {
    final InboundEvent inboundEvent = eventInput.read();
    if (inboundEvent == null) {
        // connection has been closed
        break;
    }
    System.out.println(inboundEvent.getName() + "; "
        + inboundEvent.readData(String.class));
}

In this example, a client connects to the server where the SseResource from the Example 14.2, “Simple SSE resource method” is deployed. At first, a new JAX-RS/Jersey client instance is created with a SseFeature registered. Then a WebTarget instance is retrieved from the client and is used to invoke a HTTP request. The returned response entity is directly read as a EventInput Java type, which is an extension of Jersey ChunkedInput that provides generic support for consuming chunked message payloads. The code in the example then process starts a loop to process the inbound SSE events read from the eventInput response stream. Each chunk read from the input is a InboundEvent. The method InboundEvent.readData(Class) provides a way for the client to indicate what Java type should be used for the event data de-serialization. In our example, individual events are de-serialized as String Java type instances. This method internally finds and executes a proper MessageBodyReader<T> which is the used to do the actual de-serialization. This is similar to reading an entity from the Response by readEntity(Class). The method readData can also throw a ProcessingException.

The null check on inboundEvent is necessary to make sure that the chunk was properly read and connection has not been closed by the server. Once the connection is closed, the loop terminates and the program completes execution. The client code produces the following console output:

message-to-client; Hello world 0!
message-to-client; Hello world 1!
message-to-client; Hello world 2!
message-to-client; Hello world 3!
message-to-client; Hello world 4!
message-to-client; Hello world 5!
message-to-client; Hello world 6!
message-to-client; Hello world 7!
message-to-client; Hello world 8!
message-to-client; Hello world 9!

The main Jersey SSE client API component used to read SSE events asynchronously is EventSource. The usage of the EventSource is shown on the following example.

Client client = ClientBuilder.newBuilder()
                        .register(SseFeature.class).build();
WebTarget target = client.target("http://example.com/events");
EventSource eventSource = EventSource.target(target).build();
EventListener listener = new EventListener() {
        @Override
        public void onEvent(InboundEvent inboundEvent) {
            System.out.println(inboundEvent.getName() + "; "
                    + inboundEvent.readData(String.class));
        }
    };
eventSource.register(listener, "message-to-client");
eventSource.open();
...
eventSource.close();

In this example, the client code again connects to the server where the SseResource from the Example 14.2, “Simple SSE resource method” is deployed. The Client instance is again created and initialized with SseFeature. Then the WebTarget is built. In this case a request to the web target is not made directly in the code, instead, the web target instance is used to initialize a new EventSource.Builder instance that is used to build a new EventSource. The choice of build() method is important, as it tells the EventSource.Builder to create a new EventSource that is not automatically connected to the target. The connection is established only later by manually invoking the eventSource.open() method. A custom EventListener implementation is used to listen to and process incoming SSE events. The method readData(Class) says that the event data should be de-serialized from a received InboundEvent instance into a String Java type. This method call internally executes MessageBodyReader<T> which de-serializes the event data. This is similar to reading an entity from the Response by readEntity(Class). The method readData can throw a ProcessingException.

The custom event source listener is registered in the event source via EventSource.register(EventListener, String) method. The next method arguments define the names of the events to receive and can be omitted. If names are defined, the listener will be associated with the named events and will only invoked for events with a name from the set of defined event names. It will not be invoked for events with any other name or for events without a name.

After a connection to the server is opened by calling the open() method on the event source, the eventSource starts listening to events. When an event named "message-to-client" comes, the listener will be executed by the event source. If any other event comes (with a name different from "message-to-client"), the registered listener is not invoked. Once the client is done with processing and does not want to receive events anymore, it closes the connection by calling the close() method on the event source.

The listener from the example above will print the following output:

message-to-client; Hello world 0!
message-to-client; Hello world 1!
message-to-client; Hello world 2!
message-to-client; Hello world 3!
message-to-client; Hello world 4!
message-to-client; Hello world 5!
message-to-client; Hello world 6!
message-to-client; Hello world 7!
message-to-client; Hello world 8!
message-to-client; Hello world 9!

When browsing through the Jersey SSE API documentation, you may have noticed that the EventSource implements EventListener and provides an empty implementation for the onEvent(InboundEvent inboundEvent) listener method. This adds more flexibility to the Jersey client-side SSE API. Instead of defining and registering a separate event listener, in simple scenarios you can also choose to derive directly from the EventSource and override the empty listener method to handle the incoming events. This programming model is shown in the following example:

Client client = ClientBuilder.newBuilder()
                        .register(SseFeature.class).build();
WebTarget target = client.target("http://example.com/events");
EventSource eventSource = new EventSource(target) {
    @Override
    public void onEvent(InboundEvent inboundEvent) {
        if ("message-to-client".equals(inboundEvent.getName())) {
            System.out.println(inboundEvent.getName() + "; "
                    + inboundEvent.readData(String.class));
        }
    }
};
...
eventSource.close();

The code above is very similar to the code in Example 14.4, “Registering EventListener with EventSource”. In this example however, the EventSource is constructed directly using a single-parameter constructor. This way, the connection to the SSE endpoint is by default automatically opened at the event source creation. The implementation of the EventListener has been moved into the overridden EventSource.onEvent(...) method. However, this time, the listener method will be executed for all events - unnamed as well as with any name. Therefore the code checks the name whether it is an event with the name "message-to-client" that we want to handle. Note that you can still register additional EventListeners later on. The overridden method on the event source allows you to handle messages even when no additional listeners are registered yet.