SSE 接続

リアルタイムデータは Server-Sent Events（SSE）で送ります。クライアントが GET を 1 回送ると、サーバーは接続を維持してイベントを送り続けます。1 本の接続で複数のデータ流（topic）を同時に購読できます。各データは gzip で圧縮したあと base64 で符号化されているので、クライアントは 1 件ずつ復号します。

利用できる topic

topic と単一 topic のパス	内容	権限	ライセンス
`trem.rts.v1` `/api/v1/trem/sse/rts`	TREM 観測点のリアルタイムスナップショット（JSON）。毎秒 2 件	`trem.rts`	非商用
`trem.intensity.v1` `/api/v1/trem/sse/intensity`	TREM 震度速報（XML）	`trem.intensity`	非商用
`eq.report.v1` `/api/v1/trem/sse/report`	中央気象署の地震情報（JSON）	`eq.report`	商用可

TREM データは非公式の観測で、中央気象署の発表が常に優先されます。気象法により、地震の予測や地震警報の発表に使うことはできません。 注意事項をすべて読む

topic 名の末尾の .v1 はデータ形式のバージョンです。パスの /v1 は SSE 接続プロトコル（挨拶、イベント名、圧縮、ハートビート）のバージョンです。

データライセンス

データにより異なる
購読する topic によります。各 topic のデータは、それぞれの区分と表記に従います。

表記の場所、例、商用利用の申請はデータの利用条件と出典表記をご覧ください。

認証

利用できる topic の 3 つは、どれも公開 topicです。token なしでも接続でき、token の権限や購読もチェックしません。

token の付け方（いずれか 1 つ）：

header Authorization: Bearer <token>（推奨）。プレフィックスはちょうど Bearer でなければなりません（大文字と小文字を区別します）。他の形式は無視されます。
query token=<token>。header から token を取得できなかったときだけ query を見ます。

SSE は X-Api-Token header を読みません。Token の取得方法は API Token を参照してください。

token を付けるときと付けないときの違い：

状況	結果
token なし	通常どおり送り、どのアカウントにも計上しません。多重化の入口の `info` には `"anonymous": true` が付きます
有効な token	通常どおり送り、そのアカウントの利用量に計上します。接続上の各 topic について、生存 1 秒につき 1 回、その topic の権限名（リソース）に記録します。クォータを超えても公開 topic は切断されません
無効、期限切れ、または取り消し済みの token	リクエスト全体が `401 invalid token` を返し、公開 topic も送りません。計量が不要なときは token を付けないでください
token を付けたが、認証サービスが一時的に使えない	公開 topic は通常どおり送りますが、未認証として扱い（`"anonymous": true`）、計量しません

利用量とクォータの計算は利用量とクォータを参照してください。

権限が必要な topic は、token の権限、アカウントの購読、クォータの順にチェックします。通らないときの理由の文字列は切断とエラーの理由を参照してください。多重化の入口では、1 つでも topic が通れば 200 を返し、通らなかったものは info の denied に並びます。

接続

Base URL はロードバランサー入口の https://api.lb.exptech.dev です（https://api.lb-tpe1.exptech.dev または https://api.lb-khh1.exptech.dev を指定してもかまいません）。入口は 2 種類あります。

多重化の入口 /api/v1/trem/sse?topics=...：1 本の接続で複数の topic を受けます。各データには event: <topic 名> が付き、イベント名で振り分けます。新しいプログラムにはこちらを推奨します。
単一 topic のパス（上の表）：topic を 1 つだけ受けます。データイベントに event: 行はありません。ブラウザの EventSource は onmessage で受け取れます。

GEThttps://api.lb.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1

データにより異なる

サーバー

クエリパラメーター

topicsstring必須
購読する topic。カンマ区切り。繰り返して付けてもかまいません（?topics=a&topics=b）。名前は大文字と小文字を区別し、完全に同じである必要があります。前後の空白は除き、重複はまとめます。多重化の入口だけが必要とします。
modestring任意既定値 sleep
live（大文字と小文字は区別しない）は trem.rts.v1 の各スナップショットを受け取ります。省略またはそれ以外の値は sleep モードです。「モード」を参照してください。
tokenstring任意
API Token。header を設定できないクライアント（ブラウザの EventSource など）専用です。それ以外は Authorization header を使ってください。

リクエスト例

curl -N "https://api.lb.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1"

const res = await fetch("https://api.lb.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1");
if (!res.ok) throw new Error(`HTTP ${res.status}`);

// text/event-stream：逐段讀取
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
for (;;) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}

const res = await fetch("https://api.lb-tpe1.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1");
if (!res.ok) throw new Error(`HTTP ${res.status}`);

// text/event-stream：逐段讀取
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
for (;;) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}

const res = await fetch("https://api.lb-khh1.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1");
if (!res.ok) throw new Error(`HTTP ${res.status}`);

// text/event-stream：逐段讀取
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
for (;;) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}

import requests

with requests.get(
    "https://api.lb.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1",
    stream=True,
    timeout=30,
) as res:
    res.raise_for_status()
    for line in res.iter_lines(decode_unicode=True):
        if line:
            print(line)

import requests

with requests.get(
    "https://api.lb-tpe1.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1",
    stream=True,
    timeout=30,
) as res:
    res.raise_for_status()
    for line in res.iter_lines(decode_unicode=True):
        if line:
            print(line)

import requests

with requests.get(
    "https://api.lb-khh1.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1",
    stream=True,
    timeout=30,
) as res:
    res.raise_for_status()
    for line in res.iter_lines(decode_unicode=True):
        if line:
            print(line)

レスポンス

200400401

接続に成功し、text/event-stream の送信を開始します

retry: 3000

event: info
data: {"location":"lb-tpe1","topics":["trem.intensity.v1","eq.report.v1","trem.rts.v1"],"anonymous":true}

event: trem.rts.v1
data: H4sIAAAAAAAC/02OywrDIBRE...

: ping 1790619700

topics がない、または名前が不正

unknown topic: trem.rts

token が無効、期限切れ、または取り消し済み

invalid token

GET だけを受け付けます。他のメソッドは 405 を返します（本文 method not allowed）。
成功時のレスポンス header：Content-Type: text/event-stream、Cache-Control: no-cache、X-Accel-Buffering: no。
クロスオリジン（CORS）：すべてのレスポンスに Access-Control-Allow-Origin: * が付き、プリフライトは Authorization header を許可します。ブラウザは fetch() に header を付けて接続できます。
1 本の多重化接続で複数の topic に同時にデータがあるときは、trem.intensity.v1、eq.report.v1、trem.rts.v1 の順に先に送ります。

モード

mode が影響するのは trem.rts.v1 だけで、接続全体に効きます。

mode	`trem.rts.v1`	`trem.intensity.v1`、`eq.report.v1`
`sleep`（既定）	「少なくとも 1 つの観測点に `alert` がある」スナップショットだけを送ります。平常時に観測点が発火しなければ、この topic にデータはまったくありません（ハートビートだけ）	すべての件を送ります
`live`	すべてのスナップショットを送ります（毎秒 2 件）	すべての件を送ります

sleep モードで送るスナップショットの内容は live モードと同じで、すべての観測点を含みます。送らないのは、観測点が発火していないスナップショットだけです。

モードを切り替えるには接続し直す必要があります。新しいモードで 2 本目の接続を先に開き、event: info を受け取ってから古い接続を閉じると、そのあいだのデータを取りこぼしません。

配信の形式

接続に成功すると、サーバーは次の順で送ります。

retry: 3000

event: info
data: {"location":"lb-tpe1","topics":["trem.intensity.v1","eq.report.v1","trem.rts.v1"],"anonymous":true}

event: trem.rts.v1
data: H4sIAAAAAAAC/02OywrDIBRE/+Wub8XrIzHumob+RMhCgpRAia3aVci/VyF9zGrgHIbZIIVXnD1YmEP0p7yuBAg5gaW24w11jeElCCm7vIS1gA2IWiP0pdYFLGcFP26utoZ2LHi4Dv0Pqy82LYK7+5jLfBXPmvdkDlEyeYiKcf0nFtM/wY6jkKxTSIKYFigFfi4qPU37G7yO1gbLAAAA

: ping 1790619700

イベント	`data` の内容	圧縮	説明
（`retry: 3000`）	なし	いいえ	推奨する再接続間隔は 3000 ミリ秒。`EventSource` は自動で採用します
`info`	JSON	いいえ	接続成功後の最初のイベント。フィールドは下の表
topic 名（多重化の入口）／`event:` 行がない（単一 topic のパス）	base64 文字列	はい	データ。データの復号を参照
`unsubscribed`	JSON `{"topic":"...","reason":"..."}`	いいえ	多重化接続上の 1 つの topic が停止し、他の topic は続きます
`close`	プレーンテキストの理由	いいえ	サーバーがこの接続を閉じようとしています
`: ping <Unix 秒>`	（コメント行。`data` はありません）	いいえ	ハートビート。ハートビートを参照

info のフィールド：

フィールド	型	説明
`location`	`string`	この接続を提供しているサーバーノード名。例：lb-tpe1。
`topics`	`string[]`	実際に購読できた topic。送信の優先順（リクエストの順ではありません）。多重化の入口だけにあります。
`denied`	`object`	購読できなかった topic → 理由。すべて購読できたときは省略します。多重化の入口だけにあります。
`anonymous`	`boolean`	この接続でアカウントを認証していない（計量しない）とき true。そうでなければ省略します。多重化の入口だけにあります。

イベントに id: フィールドはなく、サーバーは Last-Event-ID も読みません。再接続後に中断箇所から続きません。各 topic が接続時に補送するかどうかは再接続を参照してください。

ハートビート

サーバーは 60 秒ごとにコメント行を 1 行送ります。

: ping 1790619700

数字はサーバー時刻です（Unix の秒。NTP で補正済み）。
1 本の多重化接続は、購読した topic の数に関係なく、60 秒に 1 回だけ送ります。
sleep モードでデータがないとき、ハートビートが唯一の流量です。接続が生きているかの判断に使えます。90 秒以内に 1 バイトも受け取らなければ切断とみなして再接続してください。
EventSource はコメント行を無視するので、プログラムからハートビートは見えません。自分で配信を解析するときは、これをデータとして扱わないでください。

データの復号（gzip + base64）

各データイベントの data は、元の内容 → gzip 圧縮（各件が完全な 1 つの gzip データ）→ 標準 base64（= のパディングあり）です。レスポンスに Content-Encoding: gzip は付かないので、HTTP 層は自動で展開しません。1 件ずつ自分で復号してください。info、unsubscribed、close は圧縮しません。

復号後の形式：trem.rts.v1 と eq.report.v1 は JSON、trem.intensity.v1 は XML で、どれも UTF-8 です。

ブラウザ（DecompressionStream を使用）：

async function decodeData(b64) {
  const bytes = Uint8Array.from(atob(b64), (c) => c.charCodeAt(0));
  const stream = new Blob([bytes]).stream().pipeThrough(new DecompressionStream("gzip"));
  return await new Response(stream).text();
}

Node.js：

import { gunzipSync } from "node:zlib";

function decodeData(b64) {
  return gunzipSync(Buffer.from(b64, "base64")).toString("utf8");
}

Python：

import base64
import gzip

def decode_data(b64: str) -> str:
    return gzip.decompress(base64.b64decode(b64)).decode("utf-8")

上の配信の形式の例にある trem.rts.v1 の 1 件を復号すると、次のとおりです。

{"source":"core-tnn1","ts":1790619680000,"stations":{"117825C":{"i":0.0,"pga":0.61},"11DFDBC":{"i":0.4,"pga":0.87,"alert":1},"1A50B18":{"i":3.3,"pga":4.05,"alert":1}},"eq":[[23.94,121.52,32,1790619645]]}

再接続

状況	推奨するやり方
ネットワーク切断、サーバーが接続を閉じた（`event: close` がない）	3 秒待って再接続します。連続して失敗したときは待ち時間を倍にします（3、6、12…最大 60 秒）。ランダムな遅延も加えます。`event: info` を受け取ったら待ち時間を 3 秒に戻します
90 秒以内に 1 バイトも受け取らない	自分から中断して再接続します
HTTP `400`	リクエスト自体が不正です（topic 名、`topics` の不足）。直すまで再試行しないでください
HTTP `401 invalid token`	token が無効または取り消し済みです。同じ token で再試行しないでください
HTTP `429`	少なくとも 60 秒待ってから試します
HTTP `503`	一時的なエラーです。上のバックオフに従って再試行します
`event: close`、`data: token revoked`	再接続を止め、新しい token に替えます

EventSource はネットワーク切断のとき retry: 3000 に従って自動で再接続します。ただしサーバーが 200 以外を返すと、そのまま停止します（readyState が CLOSED になる）。自分で作り直してください。event: close を受け取ったら close() を呼んでください。呼ばないと EventSource が自動で再接続します。

再接続後、切断中のデータ：

trem.rts.v1：補送しません。各件は完全なスナップショットなので、次の件を処理し続ければ足ります。
trem.intensity.v1：接続時に、直近 240 秒以内の各震度速報の最後の報を先に補送します。切断前に受け取ったものと重複することがあります。EventID と Serial で重複を除いてください。
eq.report.v1：補送しません。再接続後は GET /api/v2/eq/report で補ってください。

サーバーは各件の到達を保証しません。接続が処理しきれないとき、サーバーはこの接続のキューで最も古いデータを捨てます（trem.rts.v1 は最新の 1 件だけを残します）。1 回の書き出しが 30 秒を超えても送り出せないときは、接続をそのまま中断し、event: close はありません。

切断とエラーの理由

接続の確立に失敗したとき、サーバーは通常の HTTP ステータスコードを返し、本文はプレーンテキストです（末尾に改行があります）。

ステータス	本文（原文）	いつ起きるか
`400`	`topics required, e.g. ?topics=...`	多重化の入口に topic が 1 つもない
`400`	`unknown topic: <名称>`	topic 名が不正（大文字と小文字を区別する）
`401`	`invalid token`	無効、期限切れ、または取り消し済みの token を付けた（購読する topic に関係なく）
`401`	`unauthorized`	権限が必要な topic に token がない
`403`	`forbidden: token scope missing`	token にその topic の権限がない
`403`	`forbidden: not subscribed`	アカウントがそのリソースを購読していない
`429`	`quota exceeded`	アカウントがそのリソースのクォータを超えている。その後約 60 秒以内の再接続も `429`
`503`	`auth service unavailable`	認証サービスが一時的に使えず、かつ権限の必要な topic を要求した
`405`	`method not allowed`	`GET` ではない

多重化の入口で、要求した topic がすべて拒否されたとき、401、403、429、503 の本文のうしろに : <topic 名> が付きます。例：forbidden: not subscribed: <topic 名>。1 つでも topic が通れば 200 を返し、拒否された topic は info.denied に置かれます。理由は unauthorized、forbidden: token scope missing、forbidden: not subscribed、quota exceeded のいずれかです（topic 名は付きません）。

接続の確立後、サーバーが送ることがあるもの：

イベント	`data`（原文）	いつ起きるか
`unsubscribed`	`{"topic":"<topic 名>","reason":"quota exceeded"}`	権限が必要な topic がクォータを超えた。その topic だけを止め、他の topic は続く
`close`	`quota exceeded`	最後の topic がクォータ超過で停止し、接続が閉じる
`close`	`token revoked`	token が取り消された。約 5 秒後にそのアカウントのすべての SSE 接続を閉じる（公開 topic を含む）。同じ token で再接続すると `401 invalid token` になる
（イベントなし）	—	ネットワーク切断、書き出しのタイムアウト（30 秒）、サーバーの再起動

公開 topic がクォータを超えても、unsubscribed や close は来ません。

応用例

そのまま実行できる完全なクライアントが 2 つです。復号、ハートビートのタイムアウト、バックオフ再接続、重複排除を示します。

Node.js

Node.js 18 以上（fetch 内蔵）で、パッケージのインストールは不要です。3 つの topic を同時に購読し、ハートビートのタイムアウト、バックオフ再接続、重複排除を処理します。接続後、まず直近の地震情報を補います。

// sse-client.mjs — 実行：node sse-client.mjs
// 環境変数：MODE=live ですべての RTS スナップショットを受ける。EXPTECH_TOKEN=<token> は任意。付けるとアカウントの利用量に計上される
import { gunzipSync } from "node:zlib";

const BASE = process.env.BASE ?? "https://api.lb.exptech.dev";
const REPORT_API = process.env.REPORT_API ?? "https://api.core.exptech.dev";
const TOPICS = ["trem.intensity.v1", "eq.report.v1", "trem.rts.v1"];
const MODE = process.env.MODE === "live" ? "live" : "sleep";
const TOKEN = process.env.EXPTECH_TOKEN ?? "";

const STALE_MS = 90_000; // ハートビートは 60 秒に 1 回。90 秒何もなければ再接続
const MIN_RETRY_MS = 3_000;
const MAX_RETRY_MS = 60_000;

class FatalError extends Error {} // 再試行しても無駄なエラー

let retryMs = MIN_RETRY_MS;
const intensitySerial = new Map(); // 震度速報の EventID -> 処理済みの最新 Serial
const seenReports = new Set(); // 処理済みの地震情報 ID

function decodeData(b64) {
  return gunzipSync(Buffer.from(b64, "base64")).toString("utf8");
}

function onRts(rts) {
  const ids = Object.keys(rts.stations);
  const alerted = ids.filter((id) => "alert" in rts.stations[id]);
  console.log(`[rts] ${new Date(rts.ts).toISOString()} 測站 ${ids.length} 個，觸發 ${alerted.length} 個，地震 ${rts.eq.length} 個`);
}

function onIntensity(xml) {
  const eventId = xml.match(/<Head>[\s\S]*?<EventID>(\d+)<\/EventID>/)?.[1];
  const serial = Number(xml.match(/<Serial>(\d+)<\/Serial>/)?.[1]);
  if (!eventId || serial <= (intensitySerial.get(eventId) ?? 0)) return; // 補送または重複
  intensitySerial.set(eventId, serial);
  const maxInt = xml.match(/<MaxInt>([^<]*)<\/MaxInt>/)?.[1];
  console.log(`[intensity] ${eventId} 第 ${serial} 報，最大震度 ${maxInt}`);
}

function onReport({ reason, data }) {
  if (reason === 3 && Date.now() - data.time > 86_400_000) return; // 埋め戻された古い情報
  const isNew = !seenReports.has(data.id);
  seenReports.add(data.id);
  console.log(`[report] ${isNew ? "新報告" : "更新"} ${data.id} M${data.mag} 深度 ${data.depth} km（reason ${reason}）`);
}

async function catchUpReports() {
  // eq.report.v1 は補送しない。接続後に HTTP API で直近の情報を補う
  const res = await fetch(`${REPORT_API}/api/v2/eq/report?limit=5&minIntensity=0`);
  if (!res.ok) return console.warn(`補抓地震報告失敗：HTTP ${res.status}`);
  for (const r of await res.json()) {
    if (!seenReports.has(r.id)) console.log(`[report] 補抓 ${r.id} M${r.mag}`);
    seenReports.add(r.id);
  }
}

function handleBlock(block) {
  let event = "message";
  const data = [];
  for (const line of block.split("\n")) {
    if (line.startsWith(":")) continue; // ハートビート `: ping <秒>`
    if (line.startsWith("event:")) event = line.slice(6).trim();
    else if (line.startsWith("data:")) data.push(line.slice(5).replace(/^ /, ""));
  }
  if (data.length === 0) return; // `retry: 3000` またはハートビート
  const text = data.join("\n");

  switch (event) {
    case "info":
      retryMs = MIN_RETRY_MS;
      console.log("已連線", text);
      catchUpReports().catch((err) => console.warn("補抓地震報告失敗：", err.message));
      break;
    case "unsubscribed":
      console.warn("topic 已停止：", text);
      break;
    case "close":
      if (text === "token revoked") throw new FatalError("token 已被撤銷，請更換 token");
      throw new Error(`伺服器關閉連線：${text}`);
    case "trem.rts.v1":
      onRts(JSON.parse(decodeData(text)));
      break;
    case "trem.intensity.v1":
      onIntensity(decodeData(text));
      break;
    case "eq.report.v1":
      onReport(JSON.parse(decodeData(text)));
      break;
  }
}

async function connect() {
  const url = new URL("/api/v1/trem/sse", BASE);
  url.searchParams.set("topics", TOPICS.join(","));
  if (MODE === "live") url.searchParams.set("mode", "live");

  const headers = { Accept: "text/event-stream" };
  if (TOKEN) headers.Authorization = `Bearer ${TOKEN}`;

  const ac = new AbortController();
  let timer = setTimeout(() => ac.abort(), STALE_MS);
  try {
    const res = await fetch(url, { headers, signal: ac.signal });
    if (!res.ok) {
      const body = (await res.text()).trim();
      if ([400, 401, 403, 405].includes(res.status)) throw new FatalError(`HTTP ${res.status} ${body}`);
      throw new Error(`HTTP ${res.status} ${body}`);
    }
    const decoder = new TextDecoder();
    let buf = "";
    for await (const chunk of res.body) {
      clearTimeout(timer);
      timer = setTimeout(() => ac.abort(), STALE_MS);
      buf += decoder.decode(chunk, { stream: true }).replace(/\r/g, "");
      let end;
      while ((end = buf.indexOf("\n\n")) >= 0) {
        const block = buf.slice(0, end);
        buf = buf.slice(end + 2);
        handleBlock(block);
      }
    }
  } finally {
    clearTimeout(timer);
    ac.abort();
  }
}

for (;;) {
  try {
    await connect();
    console.warn("連線已結束");
  } catch (err) {
    if (err instanceof FatalError) {
      console.error(err.message);
      process.exit(1);
    }
    console.warn("連線中斷：", err.name === "AbortError" ? "90 秒沒有收到資料" : err.message);
  }
  const wait = retryMs + Math.random() * 1000;
  retryMs = Math.min(retryMs * 2, MAX_RETRY_MS);
  console.log(`${Math.round(wait / 1000)} 秒後重連`);
  await new Promise((resolve) => setTimeout(resolve, wait));
}

ブラウザ

EventSource で公開 topic を購読します（token なし）。復号は非同期なので、Promise のキューで処理順を保ちます。

<!doctype html>
<meta charset="utf-8" />
<title>ExpTech SSE</title>
<pre id="log"></pre>
<script type="module">
  const URL_ = "https://api.lb.exptech.dev/api/v1/trem/sse?topics=trem.intensity.v1,eq.report.v1,trem.rts.v1";
  const log = (msg) => (document.getElementById("log").textContent = `${msg}\n` + document.getElementById("log").textContent);

  async function decodeData(b64) {
    const bytes = Uint8Array.from(atob(b64), (c) => c.charCodeAt(0));
    const stream = new Blob([bytes]).stream().pipeThrough(new DecompressionStream("gzip"));
    return await new Response(stream).text();
  }

  // 受け取った順に処理する（decodeData は非同期）
  let queue = Promise.resolve();
  const inOrder = (fn) => (event) => {
    queue = queue.then(() => fn(event)).catch((err) => console.error(err));
  };

  const intensitySerial = new Map();
  let es;
  let retryMs = 3000;

  function connect() {
    es = new EventSource(URL_);

    es.addEventListener("info", (e) => {
      retryMs = 3000;
      log(`已連線 ${e.data}`);
    });

    es.addEventListener("trem.rts.v1", inOrder(async (e) => {
      const rts = JSON.parse(await decodeData(e.data));
      const alerted = Object.values(rts.stations).filter((s) => "alert" in s).length;
      log(`RTS ${new Date(rts.ts).toLocaleTimeString()} 觸發測站 ${alerted} 個`);
    }));

    es.addEventListener("trem.intensity.v1", inOrder(async (e) => {
      const doc = new DOMParser().parseFromString(await decodeData(e.data), "application/xml");
      const eventId = doc.querySelector("Head > EventID").textContent;
      const serial = Number(doc.querySelector("Head > Serial").textContent);
      if (serial <= (intensitySerial.get(eventId) ?? 0)) return; // 補送または重複
      intensitySerial.set(eventId, serial);
      log(`震度速報 ${eventId} 第 ${serial} 報 最大震度 ${doc.querySelector("MaxInt").textContent}`);
    }));

    es.addEventListener("eq.report.v1", inOrder(async (e) => {
      const { reason, data } = JSON.parse(await decodeData(e.data));
      log(`地震報告 ${data.id} M${data.mag}（reason ${reason}）`);
    }));

    es.addEventListener("close", (e) => {
      log(`伺服器關閉連線：${e.data}`);
      es.close(); // 閉じないと EventSource が自動で再接続する
      if (e.data !== "token revoked") reconnectLater();
    });

    es.onerror = () => {
      // ネットワーク切断のとき readyState は CONNECTING で、EventSource が自分で再接続する。
      // サーバーがエラーを返したとき readyState は CLOSED なので、自分で作り直す。
      if (es.readyState === EventSource.CLOSED) reconnectLater();
    };
  }

  function reconnectLater() {
    const wait = retryMs + Math.random() * 1000;
    retryMs = Math.min(retryMs * 2, 60_000);
    setTimeout(connect, wait);
  }

  connect();
</script>

注意事項

複数の topic を受けるときは、多重化の入口で 1 本の接続にまとめてください。利用量の計算は別々に接続したときと同じですが、認証は 1 回、接続は 1 本で足ります。
trem.rts.v1 は live モードで毎秒 2 件、各件にすべての観測点が入り、流量が最大の topic です。地震のあいだのデータだけが必要なときは、既定の sleep モードを使ってください。
各 topic のデータ形式は trem.rts.v1、trem.intensity.v1、eq.report.v1 を参照してください。