WebSocket and SSE

Use fromCallback and fromEvent to receive events as streams. These examples keep at most 64 queued values and drop the oldest when the queue is full.

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

Both adapters remove their listeners during cleanup.

Use overflow: "error" to fail on overflow, or "drop-newest" to discard incoming values instead.

Consume and close the connection

The connection's owner closes it in finally. SSE receives until the supplied signal aborts; the WebSocket adapter also ends on a close event.

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