55 lines
1.4 KiB
TypeScript
55 lines
1.4 KiB
TypeScript
import type { GatewayWsClient } from "./server/ws-types.js";
|
|
import { MAX_BUFFERED_BYTES } from "./server-constants.js";
|
|
import { logWs, summarizeAgentEventForWsLog } from "./ws-log.js";
|
|
|
|
export function createGatewayBroadcaster(params: { clients: Set<GatewayWsClient> }) {
|
|
let seq = 0;
|
|
const broadcast = (
|
|
event: string,
|
|
payload: unknown,
|
|
opts?: {
|
|
dropIfSlow?: boolean;
|
|
stateVersion?: { presence?: number; health?: number };
|
|
},
|
|
) => {
|
|
const eventSeq = ++seq;
|
|
const frame = JSON.stringify({
|
|
type: "event",
|
|
event,
|
|
payload,
|
|
seq: eventSeq,
|
|
stateVersion: opts?.stateVersion,
|
|
});
|
|
const logMeta: Record<string, unknown> = {
|
|
event,
|
|
seq: eventSeq,
|
|
clients: params.clients.size,
|
|
dropIfSlow: opts?.dropIfSlow,
|
|
presenceVersion: opts?.stateVersion?.presence,
|
|
healthVersion: opts?.stateVersion?.health,
|
|
};
|
|
if (event === "agent") {
|
|
Object.assign(logMeta, summarizeAgentEventForWsLog(payload));
|
|
}
|
|
logWs("out", "event", logMeta);
|
|
for (const c of params.clients) {
|
|
const slow = c.socket.bufferedAmount > MAX_BUFFERED_BYTES;
|
|
if (slow && opts?.dropIfSlow) continue;
|
|
if (slow) {
|
|
try {
|
|
c.socket.close(1008, "slow consumer");
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
continue;
|
|
}
|
|
try {
|
|
c.socket.send(frame);
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
}
|
|
};
|
|
return { broadcast };
|
|
}
|