一覧WB-11-3 · ブラウザからの非同期通信

ポーリング・SSE・WebSocket

在庫10から始めて、変化のないポーリングにも要求が発生することを確認します。続いてSSEの複数イベントと再接続時のID、WebSocketの購読要求・通知・終了を観察し、通信方向とメッセージの区切りを整理します。

説明

在庫や進捗を画面へ反映する方法には、繰り返し取得するポーリングと、継続した接続で通知を受ける方法があります。SSEはHTTPのイベントストリームを受信し、WebSocketは接続上で双方がメッセージを送れます。

在庫10から始めて、変化のないポーリングにも要求が発生することを確認します。続いてSSEの複数イベントと再接続時のID、WebSocketの購読要求・通知・終了を観察し、通信方向とメッセージの区切りを整理します。

1一回目のポーリング

在庫と要求回数を取得します。

await(await fetch("/poll")現在の在庫を要求).json()返された在庫と回数

在庫は10、要求回数は1です。ポーリングはこの取得を一定間隔などで繰り返し、その時点の状態を受け取る方式です。

2変化がなくても再要求する

同じ資源をもう一度取得します。

await(await fetch("/poll")同じ資源へ再要求).json()

在庫は10のままですが要求回数は2です。変化が少ない資源でも取得要求は発生します。間隔を長くすると要求数は減りますが、変化を知るまでの待ち時間が増えます。

3在庫を一つ減らす

在庫を更新し、次の取得を待つ状態にします。

await(await fetch("/advance"在庫を更新,{method:"POST"更新操作を要求})).json()

在庫は9になりました。ポーリング側が取得しなければ、この変更はまだその取得結果には現れません。更新操作と状態取得を別に観察します。

4次の取得で変更を知る

三回目のポーリングを実行します。

await(await fetch("/poll")更新後の状態を取得).json()

在庫9、要求回数3が返ります。前回取得からの変化を次回取得で知るという流れです。更新時刻と取得時刻が異なるため、間のすべての変化を受け取れるとは限りません。

5SSEで二件を受信する

EventSourceを作り、stockイベントを二件受け取ったらcloseします。

(globalThis.receive=limit=>new Promise(resolve=>{const source=new EventSource("/sse")イベントストリームを接続,events=[];source.addEventListener("stock"stock種別を受信,e=>{events.push({id:e.lastEventIdイベントのID,data:JSON.parse(e.data)});if(events.length===limit){source.close()継続受信を終了;resolve({events,state:source.readyState});}})}),await receive(2))

idが1・2の二件を受信します。event: stockがイベント種別、data行が文字列本文、空行がイベントの区切りです。e.dataは文字列なので、ここではJSON.parseで在庫値を取り出します。close後のreadyState=2はCLOSEDです。

6イベントストリームの行構造

受信した二件を表すイベントストリームを、空行も含めて表示します。

(await(await fetch("/sse-format")送信した形式を取得).text()).split("\n")空行を含め行単位で表示

id・event・dataの各行を一組として、空行でイベントが確定します。data行が複数あれば改行で連結されます。ここでは一件ごとにJSON文字列を一行で載せています。HTTPの一回の受信やTCPセグメントの境界が、そのままイベントの区切りになるわけではありません。

7SSEの要求回数

サーバーの記録から接続時のLast-Event-IDを取得します。

(await(await fetch("/records")サーバーの受信記録).json()).sseSSE要求だけ取得

初回の要求は一件で、Last-Event-IDは空です。一回のHTTP応答の中に複数イベントを流せるため、通知一件ごとに新しい取得要求を出す必要はありません。

8切断後にIDを伴って再接続する

新しいEventSourceで三件を待ちます。サーバーは二件を送って接続を閉じます。

await receive(3)再接続を含め三件まで受信

最初の接続で1・2を受け取り、再接続後に3を受け取ります。EventSourceは接続が閉じると再接続を試み、最後のIDを送ります。サーバー側はそのIDを使って続きの通知を選べます。IDだけで業務処理が厳密に一回になるわけではなく、再送や重複への対応も必要です。

9再接続ヘッダーを照合する

これまでのSSE要求を一覧にします。

(await(await fetch("/records")全接続の記録).json()).sseLast-Event-IDの並び

最初の演習の一件に加え、新しい接続の空IDと再接続のID=2が記録されます。最後に受け取ったイベントIDと次の要求のLast-Event-IDが対応します。

10WebSocketで購読を送り通知を受ける

接続後にstockの購読を送り、二件の通知を受け取って終了します。

await new Promise((resolve,reject)=>{const socket=new WebSocket双方向接続を開始("ws://127.0.0.1:4322/ws"),events=[];socket.onopen=()=>socket.send購読メッセージを送信(JSON.stringify({subscribe:"stock"}));socket.onmessage通知を受信=e=>{events.push(JSON.parse(e.data));if(events.length===2)socket.close(1000,"done")通常終了を開始};socket.onclose=e=>resolve({events,code:e.code,clean:e.wasClean,state:socket.readyState});socket.onerror=()=>reject(Error("WebSocket failed"))})

クライアントから購読メッセージを送り、サーバーから受付と在庫6の通知を受け取ります。通常終了のコードは1000、CLOSEDのreadyStateは3です。SSEのreadyState値とは異なります。WebSocketでは接続後の送信に新しいHTTP要求を一件ずつ作る必要はありません。

11受信フレームの記録

サーバーが受け取った購読メッセージのフレーム情報を取得します。

(await(await fetch("/records")受信フレームの記録).json()).websocketopcode・mask・本文を取得

opcode=1はテキストフレームで、ブラウザからサーバーへのフレームはmask付きです。payloadには購読JSONが入り、payloadBytesはUTF-8本文の長さです。maskは暗号化ではなく、通信内容の機密性にはTLSを使うwssが関わります。

12サーバー通知のフレームを解析する

サーバーが送った在庫通知の16進バイト列からヘッダーと本文を取り出します。

(globalThis.record=(await(await fetch("/records")).json()).websocket[0],globalThis.bytes=Uint8Array.from16進列をバイト配列へ(record.serverFrame.match(/../g),x=>parseInt(x,16)),{hex:record.serverFrame,fin:Boolean(bytes[0]&128FINビット),reserved:(bytes[0]>>4)&7,opcode:bytes[0]&15opcodeの4ビット,mask:Boolean(bytes[1]&128),length:bytes[1]&127短い本文の長さ,text:new TextDecoder().decode本文のUTF-8を復号(bytes.slice(2))})

先頭81はFIN=1・RSV=000・opcode=1です。次のバイトはmask=0と本文長を示し、その後がUTF-8のJSONです。この短いサーバー発フレームには拡張長もmaskキーもありません。126バイト以上の本文では長さ欄の形式が変わり、クライアント発ではmaskキーも加わります。

まとめ

ポーリングは状態取得を繰り返し、更新がないときも要求が発生します。

SSEはHTTP応答内でイベントを区切り、再接続時には最後のイベントIDを送れます。

WebSocketでは接続上で双方がメッセージを送り、フレームと終了手順で通信を管理します。

操作履歴と解説を振り返れます。

ブラウザ / Fetch・通信記録

1
await(await fetch("/poll")).json()
{
  "stock": 10,
  "request": 1
}
2
await(await fetch("/poll")).json()
{
  "stock": 10,
  "request": 2
}
3
await(await fetch("/advance",{method:"POST"})).json()
{
  "stock": 9
}
4
await(await fetch("/poll")).json()
{
  "stock": 9,
  "request": 3
}
5
(globalThis.receive=limit=>new Promise(resolve=>{const source=new EventSource("/sse"),events=[];source.addEventListener("stock",e=>{events.push({id:e.lastEventId,data:JSON.parse(e.data)});if(events.length===limit){source.close();resolve({events,state:source.readyState});}})}),await receive(2))
{
  "events": [
    {
      "id": "1",
      "data": {
        "stock": 9
      }
    },
    {
      "id": "2",
      "data": {
        "stock": 8
      }
    }
  ],
  "state": 2
}
6
(await(await fetch("/sse-format")).text()).split("\n")
[
  "id: 1",
  "event: stock",
  "data: {\"stock\":9}",
  "",
  "id: 2",
  "event: stock",
  "data: {\"stock\":8}",
  "",
  ""
]
7
(await(await fetch("/records")).json()).sse
[
  {
    "lastEventId": ""
  }
]
8
await receive(3)
{
  "events": [
    {
      "id": "1",
      "data": {
        "stock": 9
      }
    },
    {
      "id": "2",
      "data": {
        "stock": 8
      }
    },
    {
      "id": "3",
      "data": {
        "stock": 7
      }
    }
  ],
  "state": 2
}
9
(await(await fetch("/records")).json()).sse
[
  {
    "lastEventId": ""
  },
  {
    "lastEventId": ""
  },
  {
    "lastEventId": "2"
  }
]
10
await new Promise((resolve,reject)=>{const socket=new WebSocket("ws://127.0.0.1:4322/ws"),events=[];socket.onopen=()=>socket.send(JSON.stringify({subscribe:"stock"}));socket.onmessage=e=>{events.push(JSON.parse(e.data));if(events.length===2)socket.close(1000,"done")};socket.onclose=e=>resolve({events,code:e.code,clean:e.wasClean,state:socket.readyState});socket.onerror=()=>reject(Error("WebSocket failed"))})
{
  "events": [
    {
      "type": "subscribed",
      "topic": "stock"
    },
    {
      "type": "stock",
      "stock": 6
    }
  ],
  "code": 1000,
  "clean": true,
  "state": 3
}
11
(await(await fetch("/records")).json()).websocket
[
  {
    "opcode": 1,
    "masked": true,
    "payload": "{\"subscribe\":\"stock\"}",
    "payloadBytes": 21,
    "serverFrame": "811a7b2274797065223a2273746f636b222c2273746f636b223a367d"
  }
]
12
(globalThis.record=(await(await fetch("/records")).json()).websocket[0],globalThis.bytes=Uint8Array.from(record.serverFrame.match(/../g),x=>parseInt(x,16)),{hex:record.serverFrame,fin:Boolean(bytes[0]&128),reserved:(bytes[0]>>4)&7,opcode:bytes[0]&15,mask:Boolean(bytes[1]&128),length:bytes[1]&127,text:new TextDecoder().decode(bytes.slice(2))})
{
  "hex": "811a7b2274797065223a2273746f636b222c2273746f636b223a367d",
  "fin": true,
  "reserved": 0,
  "opcode": 1,
  "mask": false,
  "length": 26,
  "text": "{\"type\":\"stock\",\"stock\":6}"
}

演習完了