// The live feed of `GET /api/messages/stream`, one `SwarmMessage` per event. // The stream has no event id, so a reconnect sends the history again from the // start; the list therefore empties each time the connection opens, and the // replay refills it. Not `useSwarmTermStream`: that one keeps its rows across // a reconnect, which here would list every message twice. // // Each message gets an id as it arrives, and the events that arrive within one // frame join the list in a single update, so a long replay is one render per // frame instead of one per message. import { useEffect, useRef, useState } from "preact/hooks"; import type { ArrivedMessage, SwarmMessage } from "./messages.js"; import { messagesUrl, newest, numbered } from "./messages.js"; export type ConnectionState = "connecting" | "open" | "reconnecting" | "closed"; /** The most messages the page holds; older ones drop off. */ export const MAX_MESSAGES = 2000; export function useMessageStream(sinceMinutes: number): { messages: ArrivedMessage[]; connection: ConnectionState; } { const [messages, setMessages] = useState([]); const [connection, setConnection] = useState("connecting"); // The next id to hand out; it never goes back, so an id is not reused after // a reconnect either. const nextId = useRef(0); useEffect(() => { setMessages([]); setConnection("connecting"); const es = new EventSource(messagesUrl(sinceMinutes)); let pending: SwarmMessage[] = []; let frame: number | null = null; function drop() { pending = []; if (frame !== null) cancelAnimationFrame(frame); frame = null; } function flush() { frame = null; const batch = numbered(pending, nextId.current); pending = []; nextId.current += batch.length; setMessages((prev) => newest([...prev, ...batch], MAX_MESSAGES)); } es.onopen = () => { drop(); setMessages([]); setConnection("open"); }; es.onmessage = (e) => { let m: SwarmMessage; try { m = JSON.parse(e.data) as SwarmMessage; } catch { return; } pending.push(m); if (frame === null) frame = requestAnimationFrame(flush); }; es.onerror = () => { setConnection( es.readyState === 0 /* CONNECTING */ ? "reconnecting" : "closed", ); }; return () => { drop(); es.close(); }; }, [sinceMinutes]); return { messages, connection }; }