91 lines
3.7 KiB
TypeScript
91 lines
3.7 KiB
TypeScript
// The RPC client against an in-process server that frames messages the way Effect RPC does.
|
|
import { afterEach, describe, expect, it } from "bun:test";
|
|
import type { Server, ServerWebSocket } from "bun";
|
|
|
|
import { RpcError, RpcSocket } from "../src/t3/rpc.ts";
|
|
|
|
type Frame = { _tag: string; id?: string; requestId?: string; tag?: string; payload?: unknown };
|
|
|
|
let server: Server<undefined> | undefined;
|
|
afterEach(() => server?.stop(true));
|
|
|
|
/** Starts a server whose `onRequest` decides how to answer each Request frame. */
|
|
function serve(onRequest: (ws: ServerWebSocket<undefined>, frame: Frame, received: Frame[]) => void) {
|
|
const received: Frame[] = [];
|
|
server = Bun.serve({
|
|
port: 0,
|
|
fetch: (req, srv) => (srv.upgrade(req) ? undefined : new Response("no", { status: 400 })),
|
|
websocket: {
|
|
message(ws, data) {
|
|
const frame = JSON.parse(String(data)) as Frame;
|
|
received.push(frame);
|
|
if (frame._tag === "Request") onRequest(ws, frame, received);
|
|
},
|
|
},
|
|
});
|
|
return { url: `ws://localhost:${server.port}`, received };
|
|
}
|
|
|
|
const send = (ws: ServerWebSocket<undefined>, frame: object) => ws.send(JSON.stringify(frame));
|
|
|
|
describe("RpcSocket", () => {
|
|
it("resolves a unary call with its success value", async () => {
|
|
const { url } = serve((ws, frame) =>
|
|
send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Success", value: { echoed: frame.payload } } }),
|
|
);
|
|
const socket = await RpcSocket.connect(url);
|
|
expect(await socket.call("server.probe", { a: 1 })).toEqual({ echoed: { a: 1 } });
|
|
socket.close();
|
|
});
|
|
|
|
it("rejects with the server's typed error", async () => {
|
|
const error = { _tag: "EnvironmentAuthorizationError", message: "missing scope", requiredScope: "orchestration:operate" };
|
|
const { url } = serve((ws, frame) =>
|
|
send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Failure", cause: [{ _tag: "Fail", error }] } }),
|
|
);
|
|
const socket = await RpcSocket.connect(url);
|
|
const failure = await socket.call("orchestration.dispatchCommand", {}).catch((e: unknown) => e);
|
|
expect(failure).toBeInstanceOf(RpcError);
|
|
expect((failure as RpcError).error).toEqual(error);
|
|
socket.close();
|
|
});
|
|
|
|
it("acks every chunk so the server sends the next one", async () => {
|
|
const { url, received } = serve((ws, frame) => {
|
|
// Like Effect RPC, send one chunk, then wait for its Ack before the next.
|
|
const values = [[1, 2], [3]];
|
|
const next = () => {
|
|
const value = values.shift();
|
|
if (value) send(ws, { _tag: "Chunk", requestId: frame.id, values: value });
|
|
else send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Success", value: null } });
|
|
};
|
|
const waitForAck = setInterval(() => {
|
|
const acks = received.filter((f) => f._tag === "Ack").length;
|
|
if (acks === 2 - values.length) next();
|
|
if (values.length === 0 && acks === 2) clearInterval(waitForAck);
|
|
}, 2);
|
|
next();
|
|
});
|
|
const socket = await RpcSocket.connect(url);
|
|
const seen: unknown[] = [];
|
|
await socket.stream("orchestration.subscribeShell", {}, (value) => seen.push(value)).done;
|
|
expect(seen).toEqual([1, 2, 3]);
|
|
expect(received.filter((f) => f._tag === "Ack")).toHaveLength(2);
|
|
socket.close();
|
|
});
|
|
|
|
it("fails pending calls when the socket closes", async () => {
|
|
const { url } = serve((ws) => ws.close());
|
|
const socket = await RpcSocket.connect(url);
|
|
expect(socket.call("server.probe", {})).rejects.toThrow("socket closed");
|
|
await socket.closed;
|
|
});
|
|
|
|
it("closes a silent socket", async () => {
|
|
// This server never answers Pings.
|
|
const { url } = serve(() => {});
|
|
const socket = await RpcSocket.connect(url, { pingIntervalMs: 10, silenceTimeoutMs: 40 });
|
|
await socket.closed;
|
|
});
|
|
});
|