Add a serve mode to the TS QUIC interop peer for the provider lane
cultmesh-quic-peer.ts serve opens a CultMeshQuicRealtimeProvider on the same
test fixture credential the unit tests use, prints {status, endpoint} in the
shape spawnServeProcess already expects, waits for a first peer before
broadcasting (a provider's broadcast to zero peers silently sends nothing,
same reasoning as the C# quic-realtime-serve peer's
WaitForFirstConnectionAsync), then broadcasts --frames frames at --delivery
with --interval-ms between them, stopping early on stdin EOF like the C#
peer's WatchStdinCloseAsync.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
3f546fc5cd
commit
25d749067e
1 changed files with 104 additions and 10 deletions
|
|
@ -1,15 +1,41 @@
|
|||
// The TypeScript half of the QUIC realtime interop lane: a one-shot process
|
||||
// that dials an advertised `cultmesh-state+quic://` endpoint with the native
|
||||
// connector, receives `--expect` frames from a peer (normally the C# managed
|
||||
// provider started by `cultnet-interop.test.ts`'s
|
||||
// "CultMesh QUIC realtime" test), and prints them as one JSON line. It never
|
||||
// imports `cultmesh-ts` itself (the harness drives processes, and the
|
||||
// dependency direction is `cultmesh-ts` -> `cultnet-ts`, never the reverse);
|
||||
// this script reaches the connector through its own package's `src/`.
|
||||
// The TypeScript half of the QUIC realtime interop lane. Two modes:
|
||||
//
|
||||
// `dial` (Cut 4): a one-shot process that dials an advertised
|
||||
// `cultmesh-state+quic://` endpoint with the native connector, receives
|
||||
// `--expect` frames from a peer (normally the C# managed provider started by
|
||||
// `cultnet-interop.test.ts`'s "CultMesh QUIC realtime" test), and prints them
|
||||
// as one JSON line.
|
||||
//
|
||||
// `serve` (Cut 5): the StreamPixels role. Opens a `CultMeshQuicRealtimeProvider`
|
||||
// listener, prints `{status: "ready", endpoint}` (the same shape the C#
|
||||
// `quic-realtime-serve` peer prints), waits for a first peer before
|
||||
// broadcasting (mirroring `QuicRealtimeServeAsync`'s
|
||||
// `WaitForFirstConnectionAsync`, since a broadcast to nobody would otherwise
|
||||
// silently no-op), then broadcasts `--frames` frames at `--delivery` with
|
||||
// `--interval-ms` between them, watching stdin for EOF as its own shutdown
|
||||
// signal like the C# peer's `WatchStdinCloseAsync`.
|
||||
//
|
||||
// Neither mode imports `cultmesh-ts` itself (the harness drives processes,
|
||||
// and the dependency direction is `cultmesh-ts` -> `cultnet-ts`, never the
|
||||
// reverse); this script reaches the provider/connector through its own
|
||||
// package's `src/`.
|
||||
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import process from "node:process";
|
||||
|
||||
import { CultMeshQuicRealtimeConnector, type CultMeshRealtimeTarget } from "../../src/realtime-quic";
|
||||
import {
|
||||
CultMeshQuicRealtimeConnector,
|
||||
CultMeshQuicRealtimeProvider,
|
||||
type CultMeshRealtimeTarget,
|
||||
} from "../../src/realtime-quic";
|
||||
import type { CultMeshRealtimeDelivery, CultMeshRealtimeFrame } from "../../src/realtime-wire";
|
||||
|
||||
// `__dirname` at runtime is `dist-test/test/interop`; fixtures are binary and
|
||||
// are never compiled/copied there, so they are read from their source
|
||||
// location, three levels up. The same self-signed test credential the unit
|
||||
// tests use (`test/fixtures/README.md`): not a production credential.
|
||||
const FIXTURE_P12 = join(__dirname, "..", "..", "..", "test", "fixtures", "quic-test.p12");
|
||||
|
||||
function parseArgs(argv: readonly string[]): Map<string, string> {
|
||||
const args = new Map<string, string>();
|
||||
|
|
@ -47,9 +73,77 @@ function writeLog(event: string, payload: Record<string, unknown>): void {
|
|||
process.stderr.write(`${JSON.stringify({ event, ...payload })}\n`);
|
||||
}
|
||||
|
||||
function watchStdinCloseAsync(): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
process.stdin.resume();
|
||||
process.stdin.once("end", resolve);
|
||||
});
|
||||
}
|
||||
|
||||
async function serveAsync(args: Map<string, string>): Promise<void> {
|
||||
const frames = optionalIntArg(args, "frames", 5);
|
||||
const deliveryArg = args.get("delivery") ?? "latest-only";
|
||||
if (deliveryArg !== "latest-only" && deliveryArg !== "reliable-ordered") {
|
||||
throw new Error(`Unsupported --delivery '${deliveryArg}'.`);
|
||||
}
|
||||
const delivery = deliveryArg as CultMeshRealtimeDelivery;
|
||||
const intervalMs = optionalIntArg(args, "interval-ms", 50);
|
||||
const port = optionalIntArg(args, "port", 0);
|
||||
|
||||
const provider = await CultMeshQuicRealtimeProvider.listen({
|
||||
host: "127.0.0.1",
|
||||
port,
|
||||
serverCertificate: { pkcs12: readFileSync(FIXTURE_P12), password: "" },
|
||||
});
|
||||
writeJsonLine({ status: "ready", endpoint: provider.advertisedEndpoint });
|
||||
|
||||
try {
|
||||
// A broadcast to zero peers is not an error, but it sends nothing:
|
||||
// matching the C# `quic-realtime-serve` peer, wait for a first
|
||||
// connection, bounded, so a peer that never shows up still exits.
|
||||
const connectDeadline = Date.now() + 30_000;
|
||||
while (provider.connectionCount === 0) {
|
||||
if (Date.now() >= connectDeadline) throw new Error("quic-realtime-serve: no client connected within 30000ms.");
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
|
||||
const stdinClosed = watchStdinCloseAsync();
|
||||
let stopped = false;
|
||||
void stdinClosed.then(() => {
|
||||
stopped = true;
|
||||
});
|
||||
for (let sequence = 1; sequence <= frames && !stopped; sequence += 1) {
|
||||
const frame: CultMeshRealtimeFrame = {
|
||||
channelId: "interop.quic-realtime",
|
||||
schemaId: "gamecult.interop.quic_realtime_frame.v0",
|
||||
bodyId: "interop:quic-realtime:frame",
|
||||
producerEpoch: 1n,
|
||||
sequence: BigInt(sequence),
|
||||
delivery,
|
||||
payload: Buffer.from(`frame-${sequence}`, "utf8"),
|
||||
};
|
||||
await provider.broadcast(frame);
|
||||
writeLog("quic-realtime-serve", { sent: sequence });
|
||||
if (intervalMs > 0) await new Promise((resolve) => setTimeout(resolve, intervalMs));
|
||||
}
|
||||
} finally {
|
||||
provider.dispose();
|
||||
// Mirrors the C# peer returning from `QuicRealtimeServeAsync`: once
|
||||
// broadcasting is done (or stdin closed early), let the process exit on
|
||||
// its own instead of keeping the event loop alive on a stdin that may
|
||||
// never close. The harness still ends/kills this process explicitly in
|
||||
// its own cleanup; this only lets a plain manual run exit cleanly too.
|
||||
process.stdin.pause();
|
||||
}
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const [mode, ...rest] = process.argv.slice(2);
|
||||
if (mode !== "dial") throw new Error("Expected mode: dial");
|
||||
if (mode === "serve") {
|
||||
await serveAsync(parseArgs(rest));
|
||||
return;
|
||||
}
|
||||
if (mode !== "dial") throw new Error("Expected mode: dial | serve");
|
||||
const args = parseArgs(rest);
|
||||
const endpoint = requireArg(args, "endpoint");
|
||||
const expect = optionalIntArg(args, "expect", 1);
|
||||
|
|
|
|||
Loading…
Reference in a new issue