JBoss.orgCommunity Documentation
JAX-RS 2.1 adds more asynchronous processing support in both the Client and the Server API. The specification adds a Reactive programming style to the Client side and Server-Sent Events (SSE) protocol support to both client and server.
The specification adds support for declaring asynchronous resource methods by
returning a CompletionStage instead of using the @Suspended
annotation.
RESTEasy supports more reactive types than the specification.
The specification defines a new type of invoker named RxInvoker, and a default implementation of this type named CompletionStageRxInvoker. CompletionStageRxInvoker implements Java 8's interface CompletionStage. This interface declares a large number of methods dedicated to managing asynchronous computations.
There is also a new rx method which is used in a similar manner to async.
SSE is part of HTML standard, currently supported by many browsers. It is a server push technology, which provides a way to establish a one-way channel to continuously send data to clients. SSE events are pushed to the client via a long-running HTTP connection. In case of lost connection, clients can retrieve missed events by setting a "Last-Event-ID" HTTP header in a new request.
SSE stream has text/event-stream media type and contains multiple SSE events. SSE event is a data structure encoded with UTF-8 and contains fields and comment. The field can be event, data, id, retry and other kinds of field will be ignored.
From JAX-RS 2.1, Server-sent Events APIs are introduced to support sending, receiving and broadcasting SSE events.
As shown in the following example, a SSE resource method has the text/event-stream produce media type and an injected context parameter SseEventSink. The injected SseEventSink is the connected SSE stream where events can be sent. Another injected context Sse is an entry point for creating and broadcasting SSE events. Here is an example to demonstrate how to send SSE events every 200ms and close the stream after a "done" event.
Example 25.1.
@GET
@Path("domains/{id}")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void startDomain(@PathParam("id") final String id, @Context SseEventSink sink @Context Sse sse)
{
ExecutorService service = (ExecutorService) servletContext
.getAttribute(ExecutorServletContextListener.TEST_EXECUTOR);
service.execute(new Thread()
{
public void run()
{
try
{
sink.send(sse.newEventBuilder().name("domain-progress")
.data(String.class, "starting domain " + id + " ...").build());
Thread.sleep(200);
sink.send(sse.newEvent("domain-progress", "50%"));
Thread.sleep(200);
sink.send(sse.newEvent("domain-progress", "60%"));
Thread.sleep(200);
sink.send(sse.newEvent("domain-progress", "70%"));
Thread.sleep(200);
sink.send(sse.newEvent("domain-progress", "99%"));
Thread.sleep(200);
sink.send(sse.newEvent("domain-progress", "Done.")).thenAccept((Object obj) -> {
sink.close();
});
}
catch (final InterruptedException e)
{
logger.error(e.getMessage(), e);
}
}
});
}
With SseBroadcaster, SSE events can be broadcasted to multiple clients simultaneously. It will iterate over all registered SseEventSinks and send events to all requested SSE Stream. An application can create a SseBroadcaster from an injected context Sse. The broadcast method on a SseBroadcaster is used to send SSE events to all registered clients. The following code snippet is an example on how to create SseBroadcaster, subscribe and broadcast events to all subscribed consumers.
Example 25.2.
@GET
@Path("/subscribe")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void subscribe(@Context SseEventSink sink) throws IOException
{
if (sink == null)
{
throw new IllegalStateException("No client connected.");
}
if (sseBroadcaster == null)
{
sseBroadcaster = sse.newBroadcaster();
}
sseBroadcaster.register(sink);
}
@POST
@Path("/broadcast")
public void broadcast(String message) throws IOException
{
if (sseBroadcaster == null)
{
sseBroadcaster = sse.newBroadcaster();
}
sseBroadcaster.broadcast(sse.newEvent(message));
}
SseEventSource is the entry point to read and process incoming SSE events. A SseEventSource instance can be initialized with a WebTarget. Once SseEventSource is created and connected to a server, registered event consumer will be invoked when an inbound event arrives. In case of errors, an exception will be passed to a registered consumer so that it can be processed. SseEventSource can automatically reconnect the server and continuously receive pushed events after the connection has been lost. SseEventSource can send lastEventId to the server by default when it is reconnected, and server may use this id to replay all missed events. But reply event is really upon on SSE resource method implementation. If the server responds HTTP 503 with a RETRY_AFTER header, SseEventSource will automatically schedule a reconnect task with this RETRY_AFTER value. The following code snippet is to create a SseEventSource and print the inbound event data value and error if it happens.
Example 25.3.
public void printEvent() throws Exception
{
WebTarget target = client.target("http://localhost:8080/service/server-sent-events"));
SseEventSource msgEventSource = SseEventSource.target(target).build();
try (SseEventSource eventSource = msgEventSource)
{
eventSource.register(event -> {
System.out.println(event.readData(String.class));
}, ex -> {
ex.printStackTrace();
});
eventSource.open();
}
}