File size: 3,699 Bytes
fc93158 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import { randomUUID } from "node:crypto";
import WebSocket from "ws";
export type GatewayReqFrame = { type: "req"; id: string; method: string; params?: unknown };
export type GatewayResFrame = {
type: "res";
id: string;
ok: boolean;
payload?: unknown;
error?: unknown;
};
export type GatewayEventFrame = { type: "event"; event: string; seq?: number; payload?: unknown };
export type GatewayFrame =
| GatewayReqFrame
| GatewayResFrame
| GatewayEventFrame
| { type: string; [key: string]: unknown };
export function createArgReader(argv = process.argv.slice(2)) {
const get = (flag: string) => {
const idx = argv.indexOf(flag);
if (idx !== -1 && idx + 1 < argv.length) {
return argv[idx + 1];
}
return undefined;
};
const has = (flag: string) => argv.includes(flag);
return { argv, get, has };
}
export function resolveGatewayUrl(urlRaw: string): URL {
const url = new URL(urlRaw.includes("://") ? urlRaw : `wss://${urlRaw}`);
if (!url.port) {
url.port = url.protocol === "wss:" ? "443" : "80";
}
return url;
}
function toText(data: WebSocket.RawData): string {
if (typeof data === "string") {
return data;
}
if (data instanceof ArrayBuffer) {
return Buffer.from(data).toString("utf8");
}
if (Array.isArray(data)) {
return Buffer.concat(data.map((chunk) => Buffer.from(chunk))).toString("utf8");
}
return Buffer.from(data as Buffer).toString("utf8");
}
export function createGatewayWsClient(params: {
url: string;
handshakeTimeoutMs?: number;
openTimeoutMs?: number;
onEvent?: (evt: GatewayEventFrame) => void;
}) {
const ws = new WebSocket(params.url, { handshakeTimeout: params.handshakeTimeoutMs ?? 8000 });
const pending = new Map<
string,
{
resolve: (res: GatewayResFrame) => void;
reject: (err: Error) => void;
timeout: ReturnType<typeof setTimeout>;
}
>();
const request = (method: string, paramsObj?: unknown, timeoutMs = 12_000) =>
new Promise<GatewayResFrame>((resolve, reject) => {
const id = randomUUID();
const frame: GatewayReqFrame = { type: "req", id, method, params: paramsObj };
const timeout = setTimeout(() => {
pending.delete(id);
reject(new Error(`timeout waiting for ${method}`));
}, timeoutMs);
pending.set(id, { resolve, reject, timeout });
ws.send(JSON.stringify(frame));
});
const waitOpen = () =>
new Promise<void>((resolve, reject) => {
const t = setTimeout(
() => reject(new Error("ws open timeout")),
params.openTimeoutMs ?? 8000,
);
ws.once("open", () => {
clearTimeout(t);
resolve();
});
ws.once("error", (err) => {
clearTimeout(t);
reject(err instanceof Error ? err : new Error(String(err)));
});
});
ws.on("message", (data) => {
const text = toText(data);
let frame: GatewayFrame | null = null;
try {
frame = JSON.parse(text) as GatewayFrame;
} catch {
return;
}
if (!frame || typeof frame !== "object" || !("type" in frame)) {
return;
}
if (frame.type === "res") {
const res = frame as GatewayResFrame;
const waiter = pending.get(res.id);
if (waiter) {
pending.delete(res.id);
clearTimeout(waiter.timeout);
waiter.resolve(res);
}
return;
}
if (frame.type === "event") {
const evt = frame as GatewayEventFrame;
params.onEvent?.(evt);
}
});
const close = () => {
for (const waiter of pending.values()) {
clearTimeout(waiter.timeout);
}
pending.clear();
ws.close();
};
return { ws, request, waitOpen, close };
}
|