WebSocket と SSE

fromCallback と fromEvent でイベントをストリームとして受信します。キューの上限は64件で、満杯になったら最も古い値を捨てます。

import { fromCallback, fromEvent } from "mogami-stream";

export function websocketMessages(socket: WebSocket, signal: AbortSignal) {
  return fromCallback<MessageEvent>(
    (observer) => {
      const message = (event: MessageEvent) => observer.next(event);
      const error = (event: Event) => observer.error(event);
      const close = () => observer.complete();
      socket.addEventListener("message", message);
      socket.addEventListener("error", error);
      socket.addEventListener("close", close);
      return () => {
        socket.removeEventListener("message", message);
        socket.removeEventListener("error", error);
        socket.removeEventListener("close", close);
      };
    },
    { capacity: 64, overflow: "drop-oldest" },
  ).iterate({ signal });
}

export function sseMessages(events: EventSource, signal: AbortSignal) {
  return fromEvent<MessageEvent>(events, "message", {
    capacity: 64,
    overflow: "drop-oldest",
    onDrop(info) {
      console.warn("Dropped SSE messages in this execution:", info.count);
    },
  }).iterate({ signal });
}

どちらのアダプタも終了時にリスナーを解除します。

満杯時に失敗させる場合は overflow: "error"、新しい値を捨てる場合は "drop-newest" を指定します。

接続を消費して閉じる

接続の所有者が finally で閉じます。SSE は渡された signal が abort されるまで受信を続け、WebSocket のアダプタは close イベントでも終了します。

export async function receiveEvents(url: string, signal: AbortSignal) {
  const connection = new EventSource(url);
  try {
    for await (const event of sseMessages(connection, signal)) {
      console.log(event.data);
    }
  } finally {
    connection.close();
  }
}