blog

フロントエンド視点で整理したSSEの導入とメッセージ欠落の解決プロセス

NEXTREE.

28 10월 2026 • 9 min read

記事に登場するコードや名前は、理解しやすいように実際のコードを簡略化・変更した例です。

はじめに

担当していたプロジェクトにAIチャットボット機能を追加することになりました。ユーザーが質問を入力すると、AIが回答を生成し、その結果をチャットボットのように画面に表示する必要がありました。サーバーはAIの処理が完了した時点で、結果を一度だけ返す構成でした。

バックエンドには、すでにFlux<ServerSentEvent<...>>を返すエンドポイントが用意されていました。SSEに初めて触れ、コードを見るのも初めてだったので、まずそれが何なのかを調べてからフロントエンドを接続しました。ところが、実際にテストしてみると完了メッセージが届きませんでした。

原因は思ったよりすぐに見つかりました。私はフロントエンドを担当し、画面とSSEの接続を先に実装していました。バックエンド担当者と一緒に症状を確認しながら、原因の範囲を絞っていきました。サーバー側の修正はバックエンド担当者が行い、私は何がなぜ変更されたのか説明を聞きながら整理しました。この記事では、その過程でSSEが実際にどのように動作するのか、また目に見える症状から出発して、どのように原因を絞り込んでいったのかを、フロントエンドの視点からまとめます。

1. SSEとは？

SSEは、サーバーが一つの接続を開いたままにし、データが発生するたびにクライアントへ送信する方式です。ブラウザーには、これを受信する EventSource というAPIが組み込まれています。

一般的なリクエストとレスポンスは、次のように終了します。

클라이언트: GET /api/data
서버: [응답 완성] → 200 OK + 본문 전체 → 연결 종료

SSEではレスポンスを終了しません。

클라이언트: GET /api/tasks/{taskId}/stream
서버: 200 OK, Content-Type: text/event-stream
서버: data: {"status":"GENERATING"}\n\n
서버: data: {"status":"GENERATING"}\n\n
서버: data: {"status":"SUCCESS"}\n\n
서버: (연결 종료)

data: の後に改行が2つ続くところまでが1つのメッセージだと、仕様で定められています。ブラウザーはこの形式を解析し、チャンクが到着するたびにmessageイベントを発生させます。

そのため、フロントエンドではリスナーを一度登録するだけで済みます。

eventSource.addEventListener("message", handleEventMessage);

このように登録しておけば、サーバーがチャンクを100個送ると、handleEventMessage も100回呼び出されます。

バックエンドのFluxにもそのまま対応します。Fluxは、複数の値を時間差で順番に流す型です。Spring WebFluxは、Fluxから値が出力されるたびに、それをdata: {...} 形式にエンコードしてレスポンスストリームに書き込みます。Fluxが値を出力する回数が、そのままmessageイベントリスナーの呼び出し回数になります。

2. 別の方式だったらどうだったか

実際、選択肢は一つしかありませんでした。バックエンドにはすでにSSEのエンドポイントがあり、私はそこに接続するだけでした。ただ、実装を進めるうちに、本当にSSEである必要があったのか、別の方式ならどうだったのかという疑問が湧き、ほかの方式と比較してみました。

ポーリングでは、短い間隔で照会APIを繰り返し呼び出します。実装が最も簡単で、どのようなサーバー環境でも動作しますが、不要なリクエストが多く、リアルタイム性はポーリングの間隔に左右されます。

ロングポーリングでは、リクエストを送信し、サーバーの状態が変わるまでレスポンスを保留します。ポーリングよりリアルタイム性は高いものの、サーバーが接続を長時間保持する必要があります。SSEと比べて明確な利点がない一方で、実装だけが煩雑になります。

WebSocketは双方向の常時接続です。チャットやリアルタイム共同作業のように、クライアントも継続的にデータを送る必要がある状況に適しています。

	ポーリング	ロングポーリング	SSE	WebSocket
方向	単方向	単方向	単方向	双方向
実装難易度	非常に簡単	中程度	簡単	比較的複雑
リアルタイム性	低い	中程度	高い	非常に高い
自動再接続	該当なし	独自実装	ブラウザーが標準サポート	独自実装

私たちのケースは、サーバーが生成した結果をクライアントに返すだけの一方向通信でした。ユーザーが質問を送信した後、クライアントからサーバーに追加で送るものはなく、サーバーも処理が終わった時点で結果を一度送信すれば十分でした。

この条件だけを見れば、ポーリングでも十分に実装できました。ただしポーリングでは、結果がいつ出るかわからないまま定期的にリクエストを送る必要があり、結果を受け取るタイミングもその間隔に左右されます。SSEなら1つの接続で待機し、結果が出たらすぐに受け取れるため、待機状態を表示した後、すぐ結果に切り替える必要があるチャットボット画面に適していました。結果として、すでに導入されていたSSEをそのまま使ったのも妥当な判断だったと思います。

3. JWT認証の問題とライブラリの選択

私たちのチャットボットは、まず質問を登録するAPIを呼び出してジョブIDを受け取り、そのIDでSSE接続を開く構成でした。

フロントエンドの実装は難しくなさそうでした。EventSourceインスタンスを作成してリスナーを登録すれば完了です。

const eventSource = new EventSource(url);

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data);
};

ところが接続してみると、すぐに認証エラーが発生しました。

ブラウザー内蔵のEventSourceはカスタムヘッダーに対応していません。コンストラクターに渡せるのはURLとwithCredentialsの2つだけです。私たちのプロジェクトでは、リクエストごとにヘッダーへJWTトークンを含めて認証していましたが、そのヘッダーを付ける方法自体がありませんでした。

調べてみると、この問題を回避できるライブラリが2つありました。

event-source-polyfillは内部的にXHRを使ってEventSourceAPIを模倣しながら、カスタムヘッダーに対応しています。APIの形式はブラウザー内蔵のEventSourceとほぼ同じなので、addEventListenerやclose()といった既存のコード構成をそのまま使えます。

@microsoft/fetch-event-sourceはポリフィルではなく、fetch()上でSSEを再実装したライブラリです。fetchが提供する機能をすべて利用できるため、POSTリクエストも可能で、レスポンスオブジェクトに直接アクセスして検証することもでき、再試行戦略を最初から最後まで自分で制御できます。タブがバックグラウンドに移ると接続を閉じ、再び表示されると再接続する機能もあります。

機能だけを見れば後者のほうが柔軟です。ただし私たちのバックエンドはGET方式で、プロジェクト内の別画面ではすでにevent-source-polyfillを使っていました。同じ目的のためにライブラリを2つ導入する理由はなく、既存のコード構成をほとんど変えずに済む点も大きかったため、前者を選びました。

import { EventSourcePolyfill } from "event-source-polyfill";

const eventSource = new EventSourcePolyfill(url, {
  headers: {
    Authorization: `Bearer ${token}`,
  },
  heartbeatTimeout: 5 * 60 * 1000,
});

eventSource.addEventListener("message", handleEventMessage);
eventSource.addEventListener("error", handleEventError);

4. 最初のメッセージ以降、応答が届かない症状

テストしてみると、最初のメッセージは正常に届きました。

eventMessage { status: "GENERATING", message: "Generating response...", ... }

しかし、それで終わりでした。AIが回答をすべて生成し終えているはずの時間がかなり過ぎても完了メッセージは届かず、そのまま5分待った末に返ってきたのは次のようなエラーでした。

EventSource failed.
Error: No activity within 300000 milliseconds. 277 chars received. Reconnecting.

heartbeatTimeoutに設定していた5分間、データがまったく届かなかったため、ライブラリが接続を切断して再接続を試みたのです。277 charsのうち、最初のメッセージが1つ届いただけだったということです。

5. 原因分析

5-1. フロントエンドの除外と繰り返されるメッセージ

まず、フロントエンドが関与する可能性を排除する必要がありました。ブラウザーでチャットボットを一度実行してから、開発者ツールでジョブIDとトークンをそのままコピーし、APIテストツールから同じエンドポイントに直接リクエストを送りました。

結果は同じでした。最初のメッセージが1つ届いただけで、そのまま止まりました。Reactもライブラリもリスナー登録も介在していないリクエストで同じ症状が出たので、フロントエンドは容疑対象から外れました。レスポンスヘッダーのContent-Typeもtext/event-streamで正常だったため、形式の問題でもありませんでした。

同じリクエストを何度か送ってみると、送信するたびにGENERATINGメッセージはすぐに届き、内容もすべて同じでした。バックエンドのコードを見ると、その理由が分かりました。

public Flux<ServerSentEvent<TaskMessage>> subscribe(AiTask task) {
    String taskId = task.getId();

    Sinks.Many<TaskMessage> channel = this.channelMap.computeIfAbsent(taskId,
            id -> Sinks.many().replay().all());

    channel.tryEmitNext(TaskMessage.of(task, GENERATING));

    return channel.asFlux().map(m -> ServerSentEvent.builder(m).build());
}

tryEmitNextはFluxを返す前に、メソッド本体で無条件に実行されていました。このメッセージはサーバーが進行状況を検知して送ったのではなく、リクエストが届いたというだけで、その場で生成されたものでした。つまり、私が送ったリクエストはイベントを受け取るのではなく、イベントを作り出していたのです。状態値もGENERATINGに固定されていたため、何度呼び出しても同じ内容になるのは当然でした。

5-2. 完了メッセージが破棄された理由

残る問題は、なぜ完了メッセージが届かないのかということでした。完了処理のコードは次のようになっていました。

private void completeTask(TaskMessage message) {
    String taskId = message.getTaskId();

    Sinks.Many<TaskMessage> channel = this.channelMap.remove(taskId);

    if (channel != null) {
        channel.tryEmitNext(message);
        channel.tryEmitComplete();
    }
}

channelMapにチャネルが登録されるのは、先ほど見たcomputeIfAbsentの箇所だけでした。しかし、これはフロントエンドがSSE接続を試みて初めて実行されるコードです。

問題が再現したのは、AI呼び出しがエラーですぐに失敗し、通常よりはるかに早く処理が終わった場合でした。完了処理がフロントエンドのSSE接続より先に実行されたのです。全体の流れを図にすると、次のようになります。

登録したことのないチャネルを取り出そうとしたので、removeはnullを返しました。条件文の中に入れなかったため、完了メッセージを送る相手がおらず、ログを一行も残さずに破棄されました。さらに、完了処理ではチャネルを取り出すとすぐにマップから削除する仕組みだったため、チャネルが存在していたとしても、遅れて接続した側はそのチャネルを見つけられませんでした。

6. 解決策 — チャネル登録の追加と取り出し方法の変更

原因は二つありました。タスクを登録するときに、そのタスクのチャネルもマップに登録する必要がありましたが、その処理が抜けていました。また、完了処理ではチャネルを取り出すとすぐにマップから削除していました。修正はバックエンド担当者が行い、三か所を変更しました。

まず、タスクを登録する段階でチャネルをマップに追加するようにしました。AI呼び出しイベントは、このメソッドが終了した後、別途非同期で発行されます。

public String registerTask(String prompt) {
    AiTask task = taskRepository.save(AiTask.create(prompt));

    this.channelMap.put(task.getId(), Sinks.many().replay().all());

    return task.getId();
}

次に、完了処理でチャネルをremoveではなくgetで取り出すように変更しました。

private void completeTask(TaskMessage message) {
    Sinks.Many<TaskMessage> channel = this.channelMap.get(message.getTaskId());

    if (channel != null) {
        channel.tryEmitNext(message);
        channel.tryEmitComplete();
    }
}

最後に、チャネルの削除処理をSSE接続側に移しました。購読者が完了シグナルまで受け取ったら、その時点でマップから削除します。すでに接続済みの購読者がいる場合に、待機メッセージを再送しないよう条件も追加しました。

public Flux<ServerSentEvent<TaskMessage>> subscribe(AiTask task) {
    String taskId = task.getId();

    Sinks.Many<TaskMessage> channel = this.channelMap.computeIfAbsent(taskId,
            id -> Sinks.many().replay().all());

    if (channel.currentSubscriberCount() == 0) {
        channel.tryEmitNext(TaskMessage.of(task, GENERATING));
    }

    return channel.asFlux()
            .doOnComplete(() -> this.channelMap.remove(taskId))
            .map(m -> ServerSentEvent.builder(m).build());
}

これで、AIの処理がSSE接続より先に終わっても、完了メッセージはマップに残っているチャネルに格納されます。フロントエンドが遅れて接続すると、同じチャネルを購読し、replay().all()がそれまでに蓄積されたメッセージを最初から再送するため、完了メッセージまで受け取れます。このとき、チャネルはすでに閉じているので、接続時に送ろうとしていた待機メッセージは配信されず、無視されます。結果を受け取った後、チャネルもマップから削除されます。

接続が先でもAIの処理完了が先でも、フロントエンドが受け取る結果は同じになりました。

おわりに

SSEそのものよりも大きな学びとして残ったのは、問題の範囲を絞り込んでいく方法でした。最初はヘッダーやライブラリの設定など、自分の側のコードを疑っていました。しかし、APIテストツールで同じリクエストを送ってみると、フロントエンドの問題ではないことがすぐに分かりました。先にそれを確認しておいたことで、バックエンド担当者と話す際にも、どこを調べればよいかが明確でした。

サーバーのコードを一緒に追っていく中で、非同期構造では処理の順序がどれほど重要かを知りました。送信側と受信側のコードにはそれぞれ問題がないように見えましたが、受信チャネルがいつ作られ、いつ削除されるかは、処理の順序によって変わっていました。フロントエンドで接続だけを担当する場合でも、サーバーがいつ、どこにメッセージを格納して送るのかまで把握しておけば、こうした問題をすばやく特定できると学びました。

Pong