JAX-RSでのサーバー送信イベント（SSE）

投稿日: 2019-10-18 2022-10-24
タグ: Apache CXF, Jakarta EE

1. 概要

Server-Sent Events（SSE）は、サーバーからクライアントへの長時間実行のモノチャネル接続を確立する方法を提供するHTTPベースの仕様です。

クライアントは、Acceptヘッダーのメディアタイプtext/event-streamを使用してSSE接続を開始します。

後で、サーバーを要求せずに自動的に更新されます。

仕様の詳細は公式仕様で確認できます。

このチュートリアルでは、SSEの新しいJAX-RS2.1実装を紹介します。

したがって、JAX-RSサーバーAPIを使用してイベントを公開する方法を見ていきます。また、JAX-RSクライアントAPIまたはcurlツールなどのHTTPクライアントのいずれかでそれらを使用する方法についても説明します。

2. SSEイベントを理解する

SSEイベントは、次のフィールドで構成されるテキストのブロックです。

イベント：イベントのタイプ。サーバーはさまざまなタイプの多くのメッセージを送信でき、クライアントは特定のタイプのみをリッスンするか、イベントタイプごとに異なる方法で処理できます。
データ：サーバーから送信されたメッセージ。同じイベントに対して多くのデータラインを持つことができます
Id：イベントのID。接続の再試行後にLast-Event-IDヘッダーを送信するために使用されます。サーバーがすでに送信されたイベントを送信するのを防ぐことができるので便利です
再試行：電流が失われたときにクライアントが新しい接続を確立する時間（ミリ秒単位）。最後に受信したIDは、Last-Event-IDヘッダーを介して自動的に送信されます
‘：‘：これはコメントであり、クライアントによって無視されます

また、2つの連続するイベントは、二重改行’ \ n \n‘で区切られます。

さらに、次の例に示すように、同じイベントのデータを多くの行に書き込むことができます。

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}

の JAX RS 、SSEイベントはによって抽象化されます SseEvent インターフェース、より正確には、2つのサブインターフェイスによって OutboundSseEvent と InboundSseEvent。

OutboundSseEventはサーバーAPIで使用され、送信されたイベントを設計しますが、InboundSseEventはクライアントAPIによって使用され、受信したイベントを抽象化します。

3. SSEイベントの公開

SSEイベントとは何かについて説明したので、それを構築してHTTPクライアントに送信する方法を見てみましょう。

3.1. プロジェクトの設定

JAXRSベースのMavenプロジェクトのセットアップに関するチュートリアルはすでにあります。依存関係を設定してJAXRSを開始する方法については、こちらをご覧ください。

3.2. SSEリソースメソッド

SSEリソースメソッドは、次のようなJAXRSメソッドです。

text /event-streamメディアタイプを生成できます
イベントが送信されるSseEventSinkパラメーターが挿入されています
イベントビルダーを作成するためのエントリポイントとして使用されるSseパラメーターが挿入されている場合もあります

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

したがって、クライアントは次のHTTPヘッダーを使用して最初のHTTPリクエストを行う必要があります。

Accept: text/event-stream

3.3. SSEインスタンス

SSEインスタンスは、JAXRSランタイムがインジェクションに使用できるようにするコンテキストBeanです。

これをファクトリとして使用して、次のものを作成できます。

OutboundSseEvent.Builder – を使用すると、イベントを作成してから
SseBroadcaster – を使用すると、複数のサブスクライバーにイベントをブロードキャストできます

それがどのように機能するか見てみましょう：

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

それでは、イベントビルダーに焦点を当てましょう。 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();

ご覧のとおり、ビルダーには、上記のすべてのイベントフィールドに値を設定するメソッドがあります。さらに、 mediaType（）メソッドを使用して、データフィールドのJavaオブジェクトを適切なテキスト形式にシリアル化します。

デフォルトでは、データフィールドのメディアタイプは text /plainです。したがって、 String データ型を処理するときに、明示的に指定する必要はありません。

それ以外の場合、カスタムオブジェクトを処理する場合は、メディアタイプを指定するか、カスタムMessageBodyWriterを提供する必要があります。 JAX RSランタイムは、最もよく知られているメディアタイプのMessageBodyWriterを提供します。

Sseインスタンスには、データフィールドのみ、またはタイプフィールドとデータフィールドのみを使用してイベントを作成するための2つのビルダーショートカットもあります。

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

3.4. 簡単なイベントの送信

これで、イベントを作成する方法がわかり、SSEリソースがどのように機能するかがわかりました。簡単なイベントを送信しましょう。

SseEventSink インターフェースは、単一のHTTP接続を抽象化します。 JAX-RSランタイムは、SSEリソースメソッドでのインジェクションによってのみ利用可能にすることができます。

イベントの送信は、呼び出すのと同じくらい簡単です。 SseEventSink。 送信（）。

次の例では、大量の株式更新を送信し、最終的にイベントストリームを閉じます。

@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();
}

すべてのイベントを送信した後、サーバーは close（）メソッドを明示的に呼び出すか、できれば try-with-resource、を SseEventSinkとして使用して、接続を閉じます。 は、AutoClosableインターフェースを拡張します。

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

サンプルアプリでは、次の場所にアクセスすると、これが実行されていることがわかります。

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

3.5. 放送イベント

ブロードキャストは、イベントが複数のクライアントに同時に送信されるプロセスです。これは、 SseBroadcaster APIによって実現され、次の3つの簡単なステップで実行されます。

最初に、前に示したように、挿入されたSseコンテキストからSseBroadcasterオブジェクトを作成します。

SseBroadcaster sseBroadcaster = sse.newBroadcaster();

次に、クライアントはSseイベントを受信できるようにサブスクライブする必要があります。これは通常、SseEventSinkコンテキストインスタンスが挿入されるSSEリソースメソッドで実行されます。

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

そして最後に、 broadcast（）メソッドを呼び出すことでイベントの公開をトリガーできます：

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

これにより、登録されている各SseEventSink。に同じイベントが送信されます。

放送を紹介するために、次のURLにアクセスできます。

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

次に、broadcast（）リソースメソッドを呼び出すことでブロードキャストをトリガーできます。

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

4. SSEイベントの消費

サーバーから送信されたSSEイベントを利用するには、任意のHTTPクライアントを使用できますが、このチュートリアルでは、JAXRSクライアントAPIを使用します。

4.1. SSE用のJAXRSクライアントAPI

SSEのクライアントAPIの使用を開始するには、JAXRSクライアント実装の依存関係を提供する必要があります。

ここでは、ApacheCXFクライアントの実装を使用します。

<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 このAPIの心臓部であり、 WebTarget。

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();
}

接続が確立されると、登録されたイベントコンシューマーが受信ごとに呼び出されます InboundSseEvent。

次に、 readData（）メソッドを使用して、元のデータ String：を読み取ることができます。

String data = inboundSseEvent.readData();

または、オーバーロードされたバージョンを使用して、適切なメディアタイプを使用して逆シリアル化されたJavaオブジェクトを取得できます。

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

ここでは、コンソールに着信イベントを出力する単純なイベントコンシューマーを提供しました。

5. 結論

このチュートリアルでは、JAXRS2.1でサーバー送信イベントを使用する方法に焦点を当てました。単一のクライアントにイベントを送信する方法と、複数のクライアントにイベントをブロードキャストする方法を示す例を提供しました。

最後に、JAX-RSクライアントAPIを使用してこれらのイベントを消費しました。

いつものように、このチュートリアルのコードはGithubのにあります。

getdocs

13036