Server-Sent Events (SSE) In JAX-RS – JAX-RS中的服务器发送事件（SSE）

最后修改: 2018年 8月 10日

中文/混合/英文(键盘快捷键：t)

1. Overview

1.概述

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. 

服务器发送事件（SSE）是一种基于HTTP的规范，它提供了一种建立从服务器到客户端的长期和单通道连接的方法。

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

客户端通过在Accept头中使用媒体类型text/event-stream来启动SSE连接。

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.

在本教程中，我们将介绍SSE的新JAX-RS 2.1实现。

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.

因此，我们将研究如何使用 JAX-RS 服务器 API 发布事件。此外，我们还将探讨如何通过JAX-RS客户端API或像curl工具那样的HTTP客户端来消费它们。

2. Understanding SSE Events

2.了解SSE事件

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

一个SSE事件是一个由以下字段组成的文本块。

  • 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‘.

另外，两个连续的事件之间用双换行’\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.

在JAX RS中，SSE事件由SseEvent接口抽象出来，或者更准确地说，由两个子接口OutboundSseEvent和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.

虽然OutboundSseEvent在服务器API上使用并设计了一个发送的事件，InboundSseEvent被客户端API使用并抽象出一个接收的事件。

3. Publishing SSE Events

3.发布SSE事件

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

现在我们讨论了什么是SSE事件，让我们看看如何建立和发送它到一个HTTP客户端。

3.1. Project Setup

3.1.项目设置

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.

我们已经有了关于设置基于JAX RS的Maven项目的教程。请看一下那里，看看如何设置依赖关系并开始使用JAX RS。

3.2. SSE Resource Method

3.2.上交所资源法

An SSE Resource method is a JAX RS method that:

一个SSE资源方法是一个JAX RS方法，它。

  • 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:

因此，客户端应该发出第一个HTTP请求，并带有以下HTTP头。

Accept: text/event-stream

3.3. The SSE Instance

3.3.SSE实例

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

SSE 实例是 JAX RS Runtime 将提供的用于注入的上下文 Bean。

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.Builder负责创建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.

正如我们所看到的，构建器具有为上述所有事件字段设置值的方法。此外，mediaType() 方法被用来将数据字段的Java对象序列化为合适的文本格式。

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.

默认情况下，数据域的媒体类型是text/plain。因此，在处理String数据类型时，不需要明确指定它。

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.

否则，如果我们想处理一个自定义对象，我们需要指定媒体类型或提供一个自定义的MessageBodyWriter。 JAX RS Runtime为大多数已知的媒体类型提供MessageBodyWriter。

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

Sse实例也有两个构建器快捷键，用于创建一个只有数据字段的事件，或类型和数据字段。

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

3.4. Sending Simple Event

3.4.发送简单事件

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

现在我们知道了如何建立事件，也了解了SSE资源的工作原理。让我们来发送一个简单的事件。

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

SseEventSink接口抽象了一个单一的 HTTP 连接。JAX-RS 运行时只能通过在 SSE 资源方法中的注入使其可用。

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

发送一个事件就像调用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:

在发送完所有的事件后，服务器通过显式调用close()方法或最好使用try-with-resource,来关闭连接，因为SseEventSink扩展了AutoClosable接口。

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

3.5.广播事件

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:

广播是将事件同时发送至多个客户端的过程。这是由SseBroadcaster API完成的，它通过三个简单的步骤完成。

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

首先，我们从一个注入的Sse上下文中创建SseBroadcaster对象，如前所示。

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:

然后，客户应该订阅，以便能够接收SSE事件。这通常是在SSE资源方法中完成的，其中注入了一个SseEventSink上下文实例。

@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:

最后，我们可以通过调用broadcast()方法触发事件发布。

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

This will send the same event to each registered SseEventSink.

这将向每个注册的SseEventSink.发送同一事件。

To showcase the broadcasting, we can access this URL:

为了展示广播，我们可以访问这个URL。

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

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

然后我们可以通过调用broadcast()资源方法来触发广播。

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

4. Consuming SSE Events

4.消耗SSE事件

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.

为了消费服务器发送的SSE事件，我们可以使用任何HTTP客户端，但在本教程中，我们将使用JAX RS客户端的API。

4.1. JAX RS Client API for SSE

4.1.用于SSE的JAX RS客户端API

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

为了开始使用SSE的客户端API，我们需要为JAX RS客户端实现提供依赖性。

Here, we’ll use Apache CXF client implementation:

在这里，我们将使用Apache CXF客户端实现。

<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.

SseEventSource是该API的核心，它由WebTarget构建。

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

我们从监听传入的事件开始，这些事件由InboundSseEvent接口抽象出来。

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.

一旦建立了连接，注册的事件消费者将为每个收到的InboundSseEvent调用。

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

然后我们可以使用readData()方法来读取原始数据String:。

String data = inboundSseEvent.readData();

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

或者我们可以使用重载版本，使用合适的媒体类型获得反序列化的Java对象。

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

5.结论

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.

在本教程中，我们着重介绍了如何使用JAX RS 2.1中的服务器发送事件。我们提供了一个例子，展示了如何向单个客户端发送事件，以及如何向多个客户端广播事件。

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

最后，我们使用JAX-RS客户端API来消费这些事件。

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

像往常一样，本教程的代码可以在Github上找到over。