JAX-RS中的服务器推送事件（Server-Sent Events, SSE）

1. Overview

Server-Sent Events (SSE) is an HTTP based specification that provides a way to establish a long-running and mono-channel connection from the server to the client.

The client initiates the SSE connection by using the media type text/event-stream in the Accept header.

Later, it gets updated automatically without requesting the server.

We can check more details about the specification on the official spec.

In this tutorial, we’ll introduce the new JAX-RS 2.1 implementation of SSE.

Hence, we’ll look at how we can publish events with the JAX-RS Server API. Also, we’ll explore how we can consume them either by the JAX-RS Client API or just by an HTTP client like the curl tool.

2. Understanding SSE Events

An SSE Event is a block of text composed of the following fields:

Event: the event’s type. The server can send many messages of different types and the client may only listen for a particular type or can process differently each event type
Data: the message sent by the server. We can have many data lines for the same event
Id: the id of the event, used to send the Last-Event-ID header, after a connection retry. It is useful as it can prevent the server from sending already sent events
Retry: the time, in milliseconds, for the client to establish a new connection when the current is lost. The last received Id will be automatically sent through the Last-Event-ID header
‘*:*‘: this is a comment and is ignored by the client

Also, two consecutive events are separated by a double newline ‘\n\n‘.

Additionally, the data in the same event can be written in many lines as can be seen in the following example:

event: stock
id: 1
: price change
retry: 4000
data: {"dateTime":"2018-07-14T18:06:00.285","id":1,
data: "name":"GOOG","price":75.7119}

event: stock
id: 2
: price change
retry: 4000
data: {"dateTime":"2018-07-14T18:06:00.285","id":2,"name":"IBM","price":83.4611}

In JAX RS, an SSE event is abstracted by the SseEvent interface*,* or more precisely, by the two subinterfaces OutboundSseEvent and InboundSseEvent.

While the OutboundSseEvent is used on the Server API and designs a sent event, the InboundSseEvent is used by the Client API and abstracts a received event.

3. Publishing SSE Events

Now that we discussed what an SSE event is let’s see how we can build and send it to an HTTP client.

3.1. Project Setup

We already have a tutorial about setting up a JAX RS-based Maven project. Feel free to have a look there to see how to set dependencies and get started with JAX RS.

3.2. SSE Resource Method

An SSE Resource method is a JAX RS method that:

Can produce a text/event-stream media type
Has an injected SseEventSink parameter, where events are sent
May also have an injected Sse parameter which is used as an entry point to create an event builder

@GET
@Path("prices")
@Produces("text/event-stream")
public void getStockPrices(@Context SseEventSink sseEventSink, @Context Sse sse) {
    //...
}

In consequence, the client should make the first HTTP request, with the following HTTP header:

Accept: text/event-stream

3.3. The SSE Instance

An SSE instance is a context bean that the JAX RS Runtime will make available for injection.

We could use it as a factory to create:

OutboundSseEvent.Builder – allows us to create events then
SseBroadcaster – allows us to broadcast events to multiple subscribers

Let’s see how that works:

@Context
public void setSse(Sse sse) {
    this.sse = sse;
    this.eventBuilder = sse.newEventBuilder();
    this.sseBroadcaster = sse.newBroadcaster();
}

Now, let’s focus on the event builder. OutboundSseEvent.Builder is responsible for creating the OutboundSseEvent:

OutboundSseEvent sseEvent = this.eventBuilder
  .name("stock")
  .id(String.valueOf(lastEventId))
  .mediaType(MediaType.APPLICATION_JSON_TYPE)
  .data(Stock.class, stock)
  .reconnectDelay(4000)
  .comment("price change")
  .build();

As we can see, the builder has methods to set values for all event fields shown above. Additionally, the mediaType() method is used to serialize the data field Java object to a suitable text format.

By default, the media type of the data field is text/plain. Hence, it doesn’t need to be explicitly specified when dealing with the String data type.

Otherwise, if we want to handle a custom object, we need to specify the media type or to provide a custom MessageBodyWriter. The JAX RS Runtime provides MessageBodyWriters for the most known media types.

The Sse instance also has two builders shortcuts for creating an event with only the data field, or the type and data fields:

OutboundSseEvent sseEvent = sse.newEvent("cool Event");
OutboundSseEvent sseEvent = sse.newEvent("typed event", "data Event");

3.4. Sending Simple Event

Now that we know how to build events and we understand how an SSE Resource works. Let’s send a simple event.

The SseEventSink interface abstracts a single HTTP connection. The JAX-RS Runtime can make it available only through injection in the SSE resource method.

Sending an event is then as simple as invoking *SseEventSink.*send().

In the next example will send a bunch of stock updates and will eventually close the event stream:

@GET
@Path("prices")
@Produces("text/event-stream")
public void getStockPrices(@Context SseEventSink sseEventSink /*..*/) {
    int lastEventId = //..;
    while (running) {
        Stock stock = stockService.getNextTransaction(lastEventId);
        if (stock != null) {
            OutboundSseEvent sseEvent = this.eventBuilder
              .name("stock")
              .id(String.valueOf(lastEventId))
              .mediaType(MediaType.APPLICATION_JSON_TYPE)
              .data(Stock.class, stock)
              .reconnectDelay(3000)
              .comment("price change")
              .build();
            sseEventSink.send(sseEvent);
            lastEventId++;
        }
     //..
    }
    sseEventSink.close();
}

After sending all events, the server closes the connection either by explicitly invoking the close() method or, preferably, by using the try-with-resource, as the SseEventSink extends the AutoClosable interface:

try (SseEventSink sink = sseEventSink) {
    OutboundSseEvent sseEvent = //..
    sink.send(sseEvent);
}

In our sample app we can see this running if we visit:

http://localhost:9080/sse-jaxrs-server/sse.html

3.5. Broadcasting Events

Broadcasting is the process by which events are sent to multiple clients simultaneously. This is accomplished by the SseBroadcaster API, and it is done in three simple steps:

First, we create the SseBroadcaster object from an injected Sse context as shown previously:

SseBroadcaster sseBroadcaster = sse.newBroadcaster();

Then, clients should subscribe to be able to receive Sse Events. This is generally done in an SSE resource method where a SseEventSink context instance is injected:

@GET
@Path("subscribe")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void listen(@Context SseEventSink sseEventSink) {
    this.sseBroadcaster.register(sseEventSink);
}

And finally, we can trigger the event publishing by invoking the broadcast() method:

@GET
@Path("publish")
public void broadcast() {
    OutboundSseEvent sseEvent = //...;
    this.sseBroadcaster.broadcast(sseEvent);
}

This will send the same event to each registered SseEventSink.

To showcase the broadcasting, we can access this URL:

http://localhost:9080/sse-jaxrs-server/sse-broadcast.html

And then we can trigger the broadcasting by invoking the broadcast() resource method:

curl -X GET http://localhost:9080/sse-jaxrs-server/sse/stock/publish

4. Consuming SSE Events

To consume an SSE event sent by the server, we can use any HTTP client, but for this tutorial, we’ll use the JAX RS client API.

4.1. JAX RS Client API for SSE

To get started with the client API for SSE, we need to provide dependencies for JAX RS Client implementation.

Here, we’ll use Apache CXF client implementation:

<dependency>
    <groupId>org.apache.cxf</groupId>
    <artifactId>cxf-rt-rs-client</artifactId>
    <version>${cxf-version}</version>
</dependency>
<dependency>
    <groupId>org.apache.cxf</groupId>
    <artifactId>cxf-rt-rs-sse</artifactId>
    <version>${cxf-version}</version>
</dependency>

The SseEventSource is the heart of this API, and it is constructed from The WebTarget.

We start by listening for incoming events whose are abstracted by the InboundSseEvent interface:

Client client = ClientBuilder.newClient();
WebTarget target = client.target(url);
try (SseEventSource source = SseEventSource.target(target).build()) {
    source.register((inboundSseEvent) -> System.out.println(inboundSseEvent));
    source.open();
}

Once the connection established, the registered event consumer will be invoked for each received InboundSseEvent.

We can then use the readData() method to read the original data String:

String data = inboundSseEvent.readData();

Or we can use the overloaded version to get the Deserialized Java Object using the suitable media type:

Stock stock = inboundSseEvent.readData(Stock.class, MediaType.Application_Json);

Here, we just provided a simple event consumer that print the incoming event in the console.

5. Conclusion

In this tutorial, we focused on how to use the Server-Sent Events in JAX RS 2.1. We provided an example that showcases how to send events to a single client as well as how to broadcast events to multiples clients.

Finally, we consumed these events using the JAX-RS client API.

As usual, the code of this tutorial can be found over on Github.

Persistence

REST

Security