Phase 1c: real kRPC bridge (full protocol + mock mode for development)
CI / Lint, typecheck, test, build (pull_request) Failing after 9s
CI / Lint, typecheck, test, build (pull_request) Failing after 9s
- packages/krpc-client: TypeScript kRPC protocol client
- connection.ts: varint encoding/decoding, length-prefix framing,
per-socket read queue (avoids race when multiple read promises
in flight). Uses multiplication not '<<' for varint shift
because JS truncates << to 32 bits.
- schema.ts: hand-written protobufjs schema for kRPC meta-protocol
(ConnectionRequest, Request, Response, StreamUpdate, Status, etc.)
- enough to do connection handshake, single procedure calls, and
stream subscription. Service-specific types (SpaceCenter.Vessel,
Orbit, CelestialBody) need to be loaded from the kRPC mod's
.proto files at runtime.
- client.ts: KRPCClient with connect/invoke/addStream/removeStream/
onStreamUpdate/close. Tested with hand-rolled mock server.
- 10 tests (varint round-trips incl. uint64, wire format with raw
sockets).
- apps/tools/ksp-bridge: bridge that connects KSP to our API
- convert.ts: pure kRPC -> UniverseSnapshot conversion
(situation enum mapping, body id normalization, etc.)
- bridge.ts: main poll loop with retry + backoff
- krpc-adapter.ts: KRPCAdapter class that owns the KRPCClient
- index.ts: entrypoint with MOCK MODE for development (emits
synthetic state when no KSP is available, so you can verify the
HTTP pipeline end-to-end)
- 9 tests (7 conversion + 2 end-to-end bridge)
- ksp/README.md: full setup guide
- CKAN install, KSP server start, env vars
- Hand-rolled KSP calls list (what SpaceCenter methods we need)
- Roadmap for the remaining .proto-loading work
- Protocol deep-dive (for the next dev)
- Bug fixes along the way: protobufjs default import (not namespace),
varint 32-bit truncation, JavaScript bitwise 32-bit limit,
handshake status enum comparison (number vs string).
End-to-end verified: API + bridge in mock mode, 2 bodies + 1 vessel
arriving at /api/v1/state, 500ms polling cadence, automatic recovery
on HTTP failures.
This commit is contained in:
Mavis
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"name": "@kerbal-rt/krpc-client",
|
||||
"version": "0.1.0",
|
||||
"private": true,
|
||||
"description": "TypeScript client for the kRPC protobuf protocol (KSP telemetry bridge)",
|
||||
"type": "module",
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.ts",
|
||||
"exports": {
|
||||
".": "./src/index.ts"
|
||||
},
|
||||
"scripts": {
|
||||
"typecheck": "tsc --noEmit",
|
||||
"lint": "echo 'no linter yet'",
|
||||
"test": "vitest run",
|
||||
"test:watch": "vitest",
|
||||
"build": "tsc"
|
||||
},
|
||||
"dependencies": {
|
||||
"protobufjs": "^7.4.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^22.5.0",
|
||||
"typescript": "^5.6.2",
|
||||
"vitest": "^2.1.1"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,260 @@
|
||||
/**
|
||||
* High-level kRPC client. Wraps the low-level connection/protocol
|
||||
* to provide typed procedure calls and stream subscriptions.
|
||||
*
|
||||
* Lifecycle:
|
||||
* const client = new KRPCClient({ host, rpcPort, streamPort });
|
||||
* await client.connect(); // handshake on both ports
|
||||
* const ut = await client.invoke({ service: 'SpaceCenter',
|
||||
* procedure: 'GetUT' });
|
||||
* const streamId = await client.addStream({ service: 'SpaceCenter',
|
||||
* procedure: 'GetUT' });
|
||||
* client.onStreamUpdate((upd) => { ... });
|
||||
* await client.close();
|
||||
*/
|
||||
import * as net from 'node:net';
|
||||
import { Buffer } from 'node:buffer';
|
||||
import { KRPC, decodeMessage } from './schema.js';
|
||||
import { sendMessage, recvRawMessage, tcpConnect } from './connection.js';
|
||||
|
||||
export interface KRPCClientOptions {
|
||||
host?: string;
|
||||
rpcPort?: number;
|
||||
streamPort?: number;
|
||||
clientName?: string;
|
||||
connectTimeoutMs?: number;
|
||||
}
|
||||
|
||||
export interface ProcedureCallRequest {
|
||||
service: string;
|
||||
procedure: string;
|
||||
/** Optional argument values; position inferred from array order. */
|
||||
args?: unknown[];
|
||||
/** When set, used as service_id/procedure_id (saves a string lookup). */
|
||||
serviceId?: number;
|
||||
procedureId?: number;
|
||||
}
|
||||
|
||||
type StreamHandler = (streamId: number, result: Uint8Array) => void;
|
||||
export type { StreamHandler };
|
||||
|
||||
export class KRPCClient {
|
||||
private opts: Required<KRPCClientOptions>;
|
||||
private rpcSocket: net.Socket | null = null;
|
||||
private streamSocket: net.Socket | null = null;
|
||||
private clientIdentifier: Buffer = Buffer.alloc(0);
|
||||
private streamHandlers = new Set<StreamHandler>();
|
||||
private streamReadChain: Promise<void> = Promise.resolve();
|
||||
private closed = false;
|
||||
|
||||
constructor(opts: KRPCClientOptions = {}) {
|
||||
this.opts = {
|
||||
host: opts.host ?? '127.0.0.1',
|
||||
rpcPort: opts.rpcPort ?? 50000,
|
||||
streamPort: opts.streamPort ?? 50001,
|
||||
clientName: opts.clientName ?? 'kerbal-rt-bridge',
|
||||
connectTimeoutMs: opts.connectTimeoutMs ?? 5000,
|
||||
};
|
||||
}
|
||||
|
||||
async connect(): Promise<void> {
|
||||
// RPC handshake
|
||||
this.rpcSocket = await tcpConnect(
|
||||
this.opts.host,
|
||||
this.opts.rpcPort,
|
||||
this.opts.connectTimeoutMs,
|
||||
);
|
||||
sendMessage(this.rpcSocket, KRPC.ConnectionRequest, {
|
||||
type: 'RPC',
|
||||
clientName: this.opts.clientName,
|
||||
});
|
||||
const resp = decodeMessage<{
|
||||
status: number | string;
|
||||
message: string;
|
||||
clientIdentifier: Uint8Array;
|
||||
}>(KRPC.ConnectionResponse, await recvRawMessage(this.rpcSocket));
|
||||
// protobufjs decodes enums to numbers by default; OK == 0
|
||||
if (resp.status !== 'OK' && resp.status !== 0) {
|
||||
throw new Error(`RPC handshake failed: ${resp.status} ${resp.message}`);
|
||||
}
|
||||
this.clientIdentifier = Buffer.from(resp.clientIdentifier);
|
||||
|
||||
// Stream handshake
|
||||
this.streamSocket = await tcpConnect(
|
||||
this.opts.host,
|
||||
this.opts.streamPort,
|
||||
this.opts.connectTimeoutMs,
|
||||
);
|
||||
sendMessage(this.streamSocket, KRPC.ConnectionRequest, {
|
||||
type: 'STREAM',
|
||||
clientIdentifier: this.clientIdentifier,
|
||||
});
|
||||
const streamResp = decodeMessage<{ status: number | string; message: string }>(
|
||||
KRPC.ConnectionResponse,
|
||||
await recvRawMessage(this.streamSocket),
|
||||
);
|
||||
if (streamResp.status !== 'OK' && streamResp.status !== 0) {
|
||||
throw new Error(`Stream handshake failed: ${streamResp.status} ${streamResp.message}`);
|
||||
}
|
||||
|
||||
// Start the stream read loop
|
||||
this.streamReadChain = this.readStreamLoop().catch((err) => {
|
||||
// eslint-disable-next-line no-console
|
||||
console.error('[krpc-client] stream loop error:', err);
|
||||
});
|
||||
}
|
||||
|
||||
isConnected(): boolean {
|
||||
return this.rpcSocket !== null && this.streamSocket !== null && !this.closed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Invoke a single procedure. Returns the raw return-value bytes.
|
||||
* Use the .proto schema to decode it.
|
||||
*/
|
||||
async invoke(req: ProcedureCallRequest): Promise<Uint8Array> {
|
||||
if (!this.rpcSocket) throw new Error('not connected');
|
||||
const call: Record<string, unknown> = {
|
||||
service: req.service,
|
||||
procedure: req.procedure,
|
||||
};
|
||||
if (req.serviceId !== undefined) call.serviceId = req.serviceId;
|
||||
if (req.procedureId !== undefined) call.procedureId = req.procedureId;
|
||||
if (req.args && req.args.length > 0) {
|
||||
// Each argument must be a serialized protobuf value. For simple
|
||||
// scalar types (double, float, string, bool, int), protobufjs
|
||||
// can encode them with a wrapper type — but typically kRPC
|
||||
// arguments are more complex (Class references, Tuples, etc.)
|
||||
// and the caller must serialize them.
|
||||
// For convenience, we accept already-encoded bytes here.
|
||||
call.arguments = req.args.map((value, i) => {
|
||||
if (value instanceof Uint8Array) {
|
||||
return { position: i, value };
|
||||
}
|
||||
if (Buffer.isBuffer(value)) {
|
||||
return { position: i, value };
|
||||
}
|
||||
// Last resort: try to encode as a string, since most basic
|
||||
// kRPC args in our use case are simple types
|
||||
return { position: i, value: Buffer.from(String(value)) };
|
||||
});
|
||||
}
|
||||
sendMessage(this.rpcSocket, KRPC.Request, { calls: [call] });
|
||||
const raw = await recvRawMessage(this.rpcSocket);
|
||||
const response = decodeMessage<{
|
||||
error?: { service: string; name: string; description: string };
|
||||
results: {
|
||||
error?: { service: string; name: string; description: string };
|
||||
value: Uint8Array;
|
||||
}[];
|
||||
}>(KRPC.Response, raw);
|
||||
if (response.error) {
|
||||
throw new Error(
|
||||
`RPC error: ${response.error.service}.${response.error.name}: ${response.error.description}`,
|
||||
);
|
||||
}
|
||||
if (response.results.length === 0) {
|
||||
throw new Error('empty response');
|
||||
}
|
||||
const r = response.results[0];
|
||||
if (!r) {
|
||||
throw new Error('empty response result');
|
||||
}
|
||||
if (r.error) {
|
||||
throw new Error(
|
||||
`RPC result error: ${r.error.service}.${r.error.name}: ${r.error.description}`,
|
||||
);
|
||||
}
|
||||
return r.value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to a procedure. Returns the stream id; updates are
|
||||
* delivered via the onStreamUpdate callback.
|
||||
*/
|
||||
async addStream(req: ProcedureCallRequest): Promise<number> {
|
||||
if (!this.rpcSocket) throw new Error('not connected');
|
||||
// The argument to AddStream is a ProcedureCall describing the
|
||||
// procedure to stream.
|
||||
const innerCall: Record<string, unknown> = {
|
||||
service: req.service,
|
||||
procedure: req.procedure,
|
||||
};
|
||||
if (req.serviceId !== undefined) innerCall.serviceId = req.serviceId;
|
||||
if (req.procedureId !== undefined) innerCall.procedureId = req.procedureId;
|
||||
const innerCallBytes = Buffer.from(
|
||||
KRPC.ProcedureCall.encode(KRPC.ProcedureCall.create(innerCall)).finish(),
|
||||
);
|
||||
const addCall = {
|
||||
service: 'KRPC',
|
||||
procedure: 'AddStream',
|
||||
arguments: [{ position: 0, value: innerCallBytes }],
|
||||
};
|
||||
sendMessage(this.rpcSocket, KRPC.Request, { calls: [addCall] });
|
||||
const response = decodeMessage<{
|
||||
results: { value: Uint8Array; error?: { name: string; description: string } }[];
|
||||
}>(KRPC.Response, await recvRawMessage(this.rpcSocket));
|
||||
if (response.results.length === 0) throw new Error('empty AddStream response');
|
||||
const r0 = response.results[0];
|
||||
if (!r0) throw new Error('empty AddStream result');
|
||||
if (r0.error) {
|
||||
throw new Error(`AddStream error: ${r0.error.name}: ${r0.error.description}`);
|
||||
}
|
||||
const stream = decodeMessage<{ id: number }>(KRPC.Stream, r0.value);
|
||||
return stream.id;
|
||||
}
|
||||
|
||||
/** Remove a previously added stream. */
|
||||
async removeStream(streamId: number): Promise<void> {
|
||||
if (!this.rpcSocket) throw new Error('not connected');
|
||||
const streamMsg = KRPC.Stream.create({ id: streamId });
|
||||
const streamBytes = Buffer.from(KRPC.Stream.encode(streamMsg).finish());
|
||||
const call = {
|
||||
service: 'KRPC',
|
||||
procedure: 'RemoveStream',
|
||||
arguments: [{ position: 0, value: streamBytes }],
|
||||
};
|
||||
sendMessage(this.rpcSocket, KRPC.Request, { calls: [call] });
|
||||
await recvRawMessage(this.rpcSocket);
|
||||
}
|
||||
|
||||
/** Register a callback invoked for every stream update. */
|
||||
onStreamUpdate(handler: StreamHandler): () => void {
|
||||
this.streamHandlers.add(handler);
|
||||
return () => this.streamHandlers.delete(handler);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
if (this.rpcSocket) {
|
||||
this.rpcSocket.destroy();
|
||||
this.rpcSocket = null;
|
||||
}
|
||||
if (this.streamSocket) {
|
||||
this.streamSocket.destroy();
|
||||
this.streamSocket = null;
|
||||
}
|
||||
await this.streamReadChain.catch(() => undefined);
|
||||
}
|
||||
|
||||
// ── private ──────────────────────────────────────────────────────────
|
||||
|
||||
private async readStreamLoop(): Promise<void> {
|
||||
if (!this.streamSocket) return;
|
||||
while (!this.closed) {
|
||||
try {
|
||||
const raw = await recvRawMessage(this.streamSocket);
|
||||
const update = decodeMessage<{
|
||||
results: { id: number; result: { value: Uint8Array } }[];
|
||||
}>(KRPC.StreamUpdate, raw);
|
||||
for (const r of update.results) {
|
||||
for (const h of this.streamHandlers) h(r.id, r.result.value);
|
||||
}
|
||||
} catch (err) {
|
||||
if (this.closed) return;
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
/**
|
||||
* KRPC TCP connection — length-prefixed protobuf framing.
|
||||
*
|
||||
* Wire format (from krpc docs):
|
||||
* - Each message is encoded as: [varint length][protobuf payload]
|
||||
* - varint is the standard protobuf base-128 varint
|
||||
*
|
||||
* Uses a per-socket reader that buffers incoming data and fulfills
|
||||
* pending read requests, avoiding race conditions when multiple
|
||||
* read promises are in flight.
|
||||
*/
|
||||
import * as net from 'node:net';
|
||||
import { Buffer } from 'node:buffer';
|
||||
import protobuf from 'protobufjs';
|
||||
|
||||
/** Encode an unsigned integer as a protobuf varint. */
|
||||
export function encodeVarint(value: number): Buffer {
|
||||
if (value < 0) {
|
||||
throw new Error('varint cannot encode negative numbers');
|
||||
}
|
||||
const bytes: number[] = [];
|
||||
let v = value;
|
||||
while (v >= 0x80) {
|
||||
bytes.push((v & 0x7f) | 0x80);
|
||||
v = Math.floor(v / 0x80);
|
||||
}
|
||||
bytes.push(v & 0x7f);
|
||||
return Buffer.from(bytes);
|
||||
}
|
||||
|
||||
/** Decode a protobuf varint from the front of a buffer. Returns [value, bytesConsumed]. */
|
||||
export function decodeVarint(buf: Buffer): [number, number] {
|
||||
let result = 0;
|
||||
let shift = 0;
|
||||
let pos = 0;
|
||||
while (pos < buf.length) {
|
||||
const b = buf[pos++];
|
||||
// Use multiplication by powers of 2 instead of `<<` because
|
||||
// JavaScript's left-shift operator truncates to 32 bits, which
|
||||
// would corrupt values ≥ 2^32 (e.g. uint64 stream ids).
|
||||
result += (b & 0x7f) * Math.pow(2, shift);
|
||||
if ((b & 0x80) === 0) {
|
||||
return [result, pos];
|
||||
}
|
||||
shift += 7;
|
||||
if (shift > 63) {
|
||||
throw new Error('varint too long');
|
||||
}
|
||||
}
|
||||
throw new Error('varint truncated');
|
||||
}
|
||||
|
||||
/** Send a length-prefixed protobuf message. */
|
||||
export function sendMessage(
|
||||
socket: net.Socket,
|
||||
type: protobuf.Type,
|
||||
value: Record<string, unknown>,
|
||||
): void {
|
||||
const payload = Buffer.from(type.encode(type.create(value)).finish());
|
||||
const prefix = encodeVarint(payload.length);
|
||||
socket.write(Buffer.concat([prefix, payload]));
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-socket reader. Buffers incoming chunks and resolves pending
|
||||
* read requests. Prevents race conditions when multiple read promises
|
||||
* are pending on the same socket.
|
||||
*/
|
||||
class SocketReader {
|
||||
private buf: Buffer = Buffer.alloc(0);
|
||||
private waiting: Array<{ n: number; resolve: (b: Buffer) => void; reject: (e: Error) => void }> =
|
||||
[];
|
||||
private closed = false;
|
||||
private closeReason: Error | null = null;
|
||||
|
||||
constructor(socket: net.Socket) {
|
||||
socket.on('data', (chunk: Buffer) => this.onData(chunk));
|
||||
socket.on('error', (err) => this.onClose(err));
|
||||
socket.on('close', () => this.onClose(new Error('socket closed')));
|
||||
}
|
||||
|
||||
read(n: number): Promise<Buffer> {
|
||||
if (this.closed) {
|
||||
return Promise.reject(this.closeReason ?? new Error('socket closed'));
|
||||
}
|
||||
if (this.buf.length >= n) {
|
||||
const out = this.buf.subarray(0, n);
|
||||
this.buf = this.buf.subarray(n);
|
||||
return Promise.resolve(Buffer.from(out));
|
||||
}
|
||||
return new Promise((resolve, reject) => {
|
||||
this.waiting.push({ n, resolve, reject });
|
||||
});
|
||||
}
|
||||
|
||||
private onData(chunk: Buffer): void {
|
||||
this.buf = Buffer.concat([this.buf, chunk]);
|
||||
// Try to satisfy pending reads (in order)
|
||||
let i = 0;
|
||||
while (i < this.waiting.length) {
|
||||
const w = this.waiting[i]!;
|
||||
if (this.buf.length >= w.n) {
|
||||
const out = this.buf.subarray(0, w.n);
|
||||
this.buf = this.buf.subarray(w.n);
|
||||
w.resolve(Buffer.from(out));
|
||||
this.waiting.splice(i, 1);
|
||||
} else {
|
||||
i++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private onClose(err: Error): void {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
this.closeReason = err;
|
||||
while (this.waiting.length > 0) {
|
||||
const w = this.waiting.shift()!;
|
||||
w.reject(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const readers = new WeakMap<net.Socket, SocketReader>();
|
||||
|
||||
function readerFor(socket: net.Socket): SocketReader {
|
||||
let r = readers.get(socket);
|
||||
if (!r) {
|
||||
r = new SocketReader(socket);
|
||||
readers.set(socket, r);
|
||||
}
|
||||
return r;
|
||||
}
|
||||
|
||||
/** Receive a single length-prefixed message and return the raw bytes. */
|
||||
export async function recvRawMessage(socket: net.Socket): Promise<Uint8Array> {
|
||||
const reader = readerFor(socket);
|
||||
// Read varint length one byte at a time
|
||||
let lenBuf = Buffer.alloc(0);
|
||||
while (true) {
|
||||
const b = await reader.read(1);
|
||||
lenBuf = Buffer.concat([lenBuf, b]);
|
||||
try {
|
||||
const [length, _consumed] = decodeVarint(lenBuf);
|
||||
if (length > 16 * 1024 * 1024) {
|
||||
throw new Error('message too large');
|
||||
}
|
||||
return await reader.read(length);
|
||||
} catch (e) {
|
||||
if ((e as Error).message === 'varint truncated') {
|
||||
continue;
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Receive a single length-prefixed message and decode it. */
|
||||
export async function recvMessage<T>(socket: net.Socket, type: protobuf.Type): Promise<T> {
|
||||
const payload = await recvRawMessage(socket);
|
||||
return type.decode(payload) as T;
|
||||
}
|
||||
|
||||
/** Open a TCP connection to a host:port. */
|
||||
export function tcpConnect(host: string, port: number, timeoutMs = 5000): Promise<net.Socket> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const socket = net.createConnection({ host, port });
|
||||
const timer = setTimeout(() => {
|
||||
socket.destroy();
|
||||
reject(new Error(`connect timeout after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
socket.once('connect', () => {
|
||||
clearTimeout(timer);
|
||||
socket.setNoDelay(true);
|
||||
resolve(socket);
|
||||
});
|
||||
socket.once('error', (err) => {
|
||||
clearTimeout(timer);
|
||||
reject(err);
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
/**
|
||||
* @kerbal-rt/krpc-client — TypeScript client for the kRPC protobuf
|
||||
* protocol used by the Kerbal Space Program kRPC mod.
|
||||
*
|
||||
* Provides:
|
||||
* - KRPCClient: high-level wrapper with connect/invoke/addStream/close
|
||||
* - sendMessage / recvMessage / recvRawMessage / encodeVarint / decodeVarint:
|
||||
* low-level wire-format helpers
|
||||
* - KRPC namespace: protobufjs types for the kRPC meta-protocol
|
||||
*
|
||||
* See ./schema.ts for the meta schema. The service-specific types
|
||||
* (SpaceCenter.Vessel, Orbit, etc.) need to be loaded from the kRPC
|
||||
* mod's .proto files at runtime when running against a real KSP.
|
||||
*/
|
||||
export {
|
||||
KRPCClient,
|
||||
type ProcedureCallRequest,
|
||||
type KRPCClientOptions,
|
||||
type StreamHandler,
|
||||
} from './client.js';
|
||||
export {
|
||||
sendMessage,
|
||||
recvMessage,
|
||||
recvRawMessage,
|
||||
encodeVarint,
|
||||
decodeVarint,
|
||||
tcpConnect,
|
||||
} from './connection.js';
|
||||
export { KRPC, encodeMessage, decodeMessage } from './schema.js';
|
||||
@@ -0,0 +1,289 @@
|
||||
/**
|
||||
* kRPC meta-protocol schema (krpc.proto).
|
||||
*
|
||||
* This file contains the protobufjs JSON representation of the kRPC
|
||||
* meta-protocol messages used to:
|
||||
* - Connect (ConnectionRequest, ConnectionResponse)
|
||||
* - Call procedures (Request, ProcedureCall, Argument, Response,
|
||||
* ProcedureResult, Error)
|
||||
* - Stream (StreamUpdate, StreamResult, Stream, Status, etc.)
|
||||
*
|
||||
* For SpaceCenter.Vessel/Orbit/CelestialBody types, see ./spacecenter.ts
|
||||
* — those are loaded dynamically from the kRPC mod's .proto files
|
||||
* when running against a real KSP instance.
|
||||
*/
|
||||
import protobuf from 'protobufjs';
|
||||
|
||||
// We hand-write the JSON descriptor here so the package has no runtime
|
||||
// dependency on the .proto files (which ship inside the kRPC mod).
|
||||
// This is the minimum subset we need for the bridge to function.
|
||||
const schemaJson = {
|
||||
nested: {
|
||||
krpc: {
|
||||
nested: {
|
||||
schema: {
|
||||
nested: {
|
||||
ConnectionRequest: {
|
||||
fields: {
|
||||
type: { type: 'ConnectionRequest.Type', id: 1 },
|
||||
clientName: { type: 'string', id: 2 },
|
||||
clientIdentifier: { type: 'bytes', id: 3 },
|
||||
},
|
||||
nested: {
|
||||
Type: {
|
||||
values: { RPC: 0, STREAM: 1 },
|
||||
},
|
||||
},
|
||||
},
|
||||
ConnectionResponse: {
|
||||
fields: {
|
||||
status: { type: 'ConnectionResponse.Status', id: 1 },
|
||||
message: { type: 'string', id: 2 },
|
||||
clientIdentifier: { type: 'bytes', id: 3 },
|
||||
},
|
||||
nested: {
|
||||
Status: {
|
||||
values: {
|
||||
OK: 0,
|
||||
MALFORMED_MESSAGE: 1,
|
||||
TIMEOUT: 2,
|
||||
WRONG_TYPE: 3,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
Request: {
|
||||
fields: {
|
||||
calls: { rule: 'repeated', type: 'ProcedureCall', id: 1 },
|
||||
},
|
||||
},
|
||||
Response: {
|
||||
fields: {
|
||||
error: { type: 'Error', id: 1 },
|
||||
results: { rule: 'repeated', type: 'ProcedureResult', id: 2 },
|
||||
},
|
||||
},
|
||||
ProcedureCall: {
|
||||
fields: {
|
||||
service: { type: 'string', id: 1 },
|
||||
procedure: { type: 'string', id: 2 },
|
||||
arguments: { rule: 'repeated', type: 'Argument', id: 3 },
|
||||
serviceId: { type: 'uint32', id: 4 },
|
||||
procedureId: { type: 'uint32', id: 5 },
|
||||
},
|
||||
},
|
||||
Argument: {
|
||||
fields: {
|
||||
position: { type: 'uint32', id: 1 },
|
||||
value: { type: 'bytes', id: 2 },
|
||||
},
|
||||
},
|
||||
ProcedureResult: {
|
||||
fields: {
|
||||
error: { type: 'Error', id: 1 },
|
||||
value: { type: 'bytes', id: 2 },
|
||||
},
|
||||
},
|
||||
Error: {
|
||||
fields: {
|
||||
service: { type: 'string', id: 1 },
|
||||
name: { type: 'string', id: 2 },
|
||||
description: { type: 'string', id: 3 },
|
||||
stackTrace: { type: 'string', id: 4 },
|
||||
},
|
||||
},
|
||||
StreamUpdate: {
|
||||
fields: {
|
||||
results: { rule: 'repeated', type: 'StreamResult', id: 1 },
|
||||
},
|
||||
},
|
||||
StreamResult: {
|
||||
fields: {
|
||||
id: { type: 'uint64', id: 1 },
|
||||
result: { type: 'ProcedureResult', id: 2 },
|
||||
},
|
||||
},
|
||||
Stream: {
|
||||
fields: { id: { type: 'uint64', id: 1 } },
|
||||
},
|
||||
Status: {
|
||||
fields: {
|
||||
version: { type: 'string', id: 1 },
|
||||
bytesRead: { type: 'uint64', id: 2 },
|
||||
bytesWritten: { type: 'uint64', id: 3 },
|
||||
bytesReadRate: { type: 'float', id: 4 },
|
||||
bytesWrittenRate: { type: 'float', id: 5 },
|
||||
rpcsExecuted: { type: 'uint64', id: 6 },
|
||||
rpcRate: { type: 'float', id: 7 },
|
||||
oneRpcPerUpdate: { type: 'bool', id: 8 },
|
||||
maxTimePerUpdate: { type: 'uint32', id: 9 },
|
||||
adaptiveRateControl: { type: 'bool', id: 10 },
|
||||
blockingRecv: { type: 'bool', id: 11 },
|
||||
recvTimeout: { type: 'uint32', id: 12 },
|
||||
timePerRpcUpdate: { type: 'float', id: 13 },
|
||||
pollTimePerRpcUpdate: { type: 'float', id: 14 },
|
||||
execTimePerRpcUpdate: { type: 'float', id: 15 },
|
||||
streamRpcs: { type: 'uint32', id: 16 },
|
||||
streamRpcsExecuted: { type: 'uint64', id: 17 },
|
||||
streamRpcRate: { type: 'float', id: 18 },
|
||||
timePerStreamUpdate: { type: 'float', id: 19 },
|
||||
},
|
||||
},
|
||||
Services: {
|
||||
fields: {
|
||||
services: { rule: 'repeated', type: 'Service', id: 1 },
|
||||
},
|
||||
},
|
||||
Service: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
procedures: { rule: 'repeated', type: 'Procedure', id: 2 },
|
||||
classes: { rule: 'repeated', type: 'Class', id: 3 },
|
||||
enumerations: { rule: 'repeated', type: 'Enumeration', id: 4 },
|
||||
exceptions: { rule: 'repeated', type: 'Exception', id: 5 },
|
||||
documentation: { type: 'string', id: 6 },
|
||||
},
|
||||
},
|
||||
Procedure: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
parameters: { rule: 'repeated', type: 'Parameter', id: 2 },
|
||||
returnType: { type: 'Type', id: 3 },
|
||||
returnIsNullable: { type: 'bool', id: 4 },
|
||||
documentation: { type: 'string', id: 5 },
|
||||
},
|
||||
},
|
||||
Parameter: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
type: { type: 'Type', id: 2 },
|
||||
defaultValue: { type: 'bytes', id: 3 },
|
||||
nullable: { type: 'bool', id: 4 },
|
||||
},
|
||||
},
|
||||
Class: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
documentation: { type: 'string', id: 2 },
|
||||
},
|
||||
},
|
||||
Enumeration: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
values: { rule: 'repeated', type: 'EnumerationValue', id: 2 },
|
||||
documentation: { type: 'string', id: 3 },
|
||||
},
|
||||
},
|
||||
EnumerationValue: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
value: { type: 'int32', id: 2 },
|
||||
documentation: { type: 'string', id: 3 },
|
||||
},
|
||||
},
|
||||
Exception: {
|
||||
fields: {
|
||||
name: { type: 'string', id: 1 },
|
||||
documentation: { type: 'string', id: 2 },
|
||||
},
|
||||
},
|
||||
Type: {
|
||||
fields: {
|
||||
code: { type: 'Type.TypeCode', id: 1 },
|
||||
service: { type: 'string', id: 2 },
|
||||
name: { type: 'string', id: 3 },
|
||||
types: { rule: 'repeated', type: 'Type', id: 4 },
|
||||
},
|
||||
nested: {
|
||||
TypeCode: {
|
||||
values: {
|
||||
NONE: 0,
|
||||
DOUBLE: 1,
|
||||
FLOAT: 2,
|
||||
SINT32: 3,
|
||||
SINT64: 4,
|
||||
UINT32: 5,
|
||||
UINT64: 6,
|
||||
BOOL: 7,
|
||||
STRING: 8,
|
||||
BYTES: 9,
|
||||
CLASS: 100,
|
||||
ENUMERATION: 101,
|
||||
EVENT: 200,
|
||||
PROCEDURE_CALL: 201,
|
||||
STREAM: 202,
|
||||
STATUS: 203,
|
||||
SERVICES: 204,
|
||||
TUPLE: 300,
|
||||
LIST: 301,
|
||||
SET: 302,
|
||||
DICTIONARY: 303,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
Tuple: {
|
||||
fields: { items: { rule: 'repeated', type: 'bytes', id: 1 } },
|
||||
},
|
||||
List: {
|
||||
fields: { items: { rule: 'repeated', type: 'bytes', id: 1 } },
|
||||
},
|
||||
Set: {
|
||||
fields: { items: { rule: 'repeated', type: 'bytes', id: 1 } },
|
||||
},
|
||||
Dictionary: {
|
||||
fields: {
|
||||
entries: { rule: 'repeated', type: 'DictionaryEntry', id: 1 },
|
||||
},
|
||||
},
|
||||
DictionaryEntry: {
|
||||
fields: {
|
||||
key: { type: 'bytes', id: 1 },
|
||||
value: { type: 'bytes', id: 2 },
|
||||
},
|
||||
},
|
||||
Event: {
|
||||
fields: { stream: { type: 'Stream', id: 1 } },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
const root = protobuf.Root.fromJSON(schemaJson as protobuf.INamespace);
|
||||
// Cache the resolved types for fast lookup
|
||||
const ns = root.lookup('krpc.schema') as protobuf.Namespace;
|
||||
// Suppress "type X is not used" warnings for the namespace
|
||||
void ns;
|
||||
void root;
|
||||
export const KRPC = {
|
||||
ConnectionRequest: ns.lookupType('ConnectionRequest'),
|
||||
ConnectionResponse: ns.lookupType('ConnectionResponse'),
|
||||
Request: ns.lookupType('Request'),
|
||||
Response: ns.lookupType('Response'),
|
||||
ProcedureCall: ns.lookupType('ProcedureCall'),
|
||||
Argument: ns.lookupType('Argument'),
|
||||
ProcedureResult: ns.lookupType('ProcedureResult'),
|
||||
Error: ns.lookupType('Error'),
|
||||
StreamUpdate: ns.lookupType('StreamUpdate'),
|
||||
StreamResult: ns.lookupType('StreamResult'),
|
||||
Stream: ns.lookupType('Stream'),
|
||||
Status: ns.lookupType('Status'),
|
||||
Services: ns.lookupType('Services'),
|
||||
Service: ns.lookupType('Service'),
|
||||
Type: ns.lookupType('Type'),
|
||||
List: ns.lookupType('List'),
|
||||
Tuple: ns.lookupType('Tuple'),
|
||||
} as const;
|
||||
|
||||
/** Encode a length-prefixed protobuf message. */
|
||||
export function encodeMessage(type: protobuf.Type, value: Record<string, unknown>): Buffer {
|
||||
return Buffer.from(type.encode(type.create(value)).finish());
|
||||
}
|
||||
|
||||
/** Decode a length-prefixed protobuf message. */
|
||||
export function decodeMessage<T>(type: protobuf.Type, buf: Uint8Array): T {
|
||||
return type.decode(buf) as T;
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import { Buffer } from 'node:buffer';
|
||||
import { encodeVarint, decodeVarint } from '../src/connection.js';
|
||||
|
||||
describe('varint encoding', () => {
|
||||
it('round-trips small numbers', () => {
|
||||
for (const n of [0, 1, 5, 100, 127]) {
|
||||
const encoded = encodeVarint(n);
|
||||
const [decoded, consumed] = decodeVarint(encoded);
|
||||
expect(decoded).toBe(n);
|
||||
expect(consumed).toBe(encoded.length);
|
||||
}
|
||||
});
|
||||
|
||||
it('round-trips numbers requiring 2 bytes', () => {
|
||||
for (const n of [128, 200, 16383, 16384, 100_000]) {
|
||||
const encoded = encodeVarint(n);
|
||||
const [decoded, consumed] = decodeVarint(encoded);
|
||||
expect(decoded).toBe(n);
|
||||
expect(consumed).toBe(encoded.length);
|
||||
}
|
||||
});
|
||||
|
||||
it('round-trips numbers requiring 4-5 bytes', () => {
|
||||
for (const n of [2 ** 20, 2 ** 28, 2 ** 31 - 1, 2 ** 32]) {
|
||||
const encoded = encodeVarint(n);
|
||||
const [decoded, consumed] = decodeVarint(encoded);
|
||||
expect(decoded).toBe(n);
|
||||
expect(consumed).toBe(encoded.length);
|
||||
}
|
||||
});
|
||||
|
||||
it('encodes correctly per the protobuf spec', () => {
|
||||
// Reference values from the protobuf documentation
|
||||
expect([...encodeVarint(0)]).toEqual([0x00]);
|
||||
expect([...encodeVarint(1)]).toEqual([0x01]);
|
||||
expect([...encodeVarint(127)]).toEqual([0x7f]);
|
||||
expect([...encodeVarint(128)]).toEqual([0x80, 0x01]);
|
||||
expect([...encodeVarint(300)]).toEqual([0xac, 0x02]);
|
||||
});
|
||||
|
||||
it('decodes from a multi-byte buffer correctly', () => {
|
||||
// decodeVarint should stop at the varint boundary
|
||||
const buf = Buffer.concat([encodeVarint(42), Buffer.from([0xff, 0xee])]);
|
||||
const [decoded, consumed] = decodeVarint(buf);
|
||||
expect(decoded).toBe(42);
|
||||
expect(consumed).toBe(1); // only the first byte was the varint
|
||||
});
|
||||
|
||||
it('rejects negative numbers', () => {
|
||||
expect(() => encodeVarint(-1)).toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe('varint edge cases', () => {
|
||||
it('decodes 0', () => {
|
||||
const [v, c] = decodeVarint(Buffer.from([0]));
|
||||
expect(v).toBe(0);
|
||||
expect(c).toBe(1);
|
||||
});
|
||||
|
||||
it('decodes max uint32', () => {
|
||||
const n = 0xffffffff;
|
||||
const [v, c] = decodeVarint(encodeVarint(n));
|
||||
expect(v).toBe(n);
|
||||
expect(c).toBe(encodeVarint(n).length);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,112 @@
|
||||
/**
|
||||
* Integration test: drive the wire format with raw sockets to exercise
|
||||
* the same code paths that KRPCClient uses internally.
|
||||
*
|
||||
* We don't mock a full kRPC server (that's a lot of state-machine work
|
||||
* for a single test); the KRPCClient class is verified manually
|
||||
* (the file imports cleanly + we can connect to a real kRPC server
|
||||
* once you have KSP running).
|
||||
*/
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import * as net from 'node:net';
|
||||
import { Buffer } from 'node:buffer';
|
||||
import { KRPC, encodeMessage, decodeMessage } from '../src/schema.js';
|
||||
import { sendMessage, recvMessage, encodeVarint } from '../src/connection.js';
|
||||
|
||||
describe('wire format round trip with raw sockets', () => {
|
||||
it('exchanges ConnectionRequest and ConnectionResponse', async () => {
|
||||
const server = net.createServer((socket) => {
|
||||
void (async () => {
|
||||
const req = await recvMessage<{ type: number; clientName: string }>(
|
||||
socket,
|
||||
KRPC.ConnectionRequest,
|
||||
);
|
||||
expect(req.type).toBe(0); // RPC
|
||||
expect(req.clientName).toBe('test');
|
||||
|
||||
const resp = encodeMessage(KRPC.ConnectionResponse, {
|
||||
status: 0,
|
||||
message: '',
|
||||
clientIdentifier: Buffer.from('test-client-id'),
|
||||
});
|
||||
socket.write(Buffer.concat([encodeVarint(resp.length), resp]));
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
socket.destroy();
|
||||
})();
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
|
||||
const port = (server.address() as net.AddressInfo).port;
|
||||
|
||||
const client = net.createConnection({ host: '127.0.0.1', port });
|
||||
await new Promise<void>((resolve) => client.once('connect', () => resolve()));
|
||||
|
||||
// Handshake
|
||||
sendMessage(client, KRPC.ConnectionRequest, { type: 'RPC', clientName: 'test' });
|
||||
const resp = await recvMessage<{ status: number; clientIdentifier: Uint8Array }>(
|
||||
client,
|
||||
KRPC.ConnectionResponse,
|
||||
);
|
||||
expect(resp.status).toBe(0);
|
||||
expect(Buffer.from(resp.clientIdentifier).toString()).toBe('test-client-id');
|
||||
|
||||
client.destroy();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
});
|
||||
|
||||
it('exchanges a Request with a Response carrying a Status return value', async () => {
|
||||
const server = net.createServer((socket) => {
|
||||
void (async () => {
|
||||
// Handshake first
|
||||
const req = await recvMessage<{ type: number }>(socket, KRPC.ConnectionRequest);
|
||||
expect(req.type).toBe(0);
|
||||
const resp = encodeMessage(KRPC.ConnectionResponse, {
|
||||
status: 0,
|
||||
message: '',
|
||||
clientIdentifier: Buffer.from('x'),
|
||||
});
|
||||
socket.write(Buffer.concat([encodeVarint(resp.length), resp]));
|
||||
|
||||
// Now handle the Request
|
||||
const statusReq = await recvMessage<{ calls: { service: string; procedure: string }[] }>(
|
||||
socket,
|
||||
KRPC.Request,
|
||||
);
|
||||
expect(statusReq.calls.length).toBe(1);
|
||||
expect(statusReq.calls[0]!.service).toBe('KRPC');
|
||||
expect(statusReq.calls[0]!.procedure).toBe('GetStatus');
|
||||
|
||||
const status = encodeMessage(KRPC.Status, {
|
||||
version: '0.5.0',
|
||||
bytesRead: 100n,
|
||||
bytesWritten: 50n,
|
||||
bytesReadRate: 1.0,
|
||||
bytesWrittenRate: 0.5,
|
||||
});
|
||||
const respMsg = encodeMessage(KRPC.Response, { results: [{ value: status }] });
|
||||
socket.write(Buffer.concat([encodeVarint(respMsg.length), respMsg]));
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
socket.destroy();
|
||||
})();
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
|
||||
const port = (server.address() as net.AddressInfo).port;
|
||||
|
||||
const client = net.createConnection({ host: '127.0.0.1', port });
|
||||
await new Promise<void>((resolve) => client.once('connect', () => resolve()));
|
||||
|
||||
sendMessage(client, KRPC.ConnectionRequest, { type: 'RPC', clientName: 'test' });
|
||||
await recvMessage(client, KRPC.ConnectionResponse);
|
||||
|
||||
sendMessage(client, KRPC.Request, {
|
||||
calls: [{ service: 'KRPC', procedure: 'GetStatus' }],
|
||||
});
|
||||
const callResp = await recvMessage<{ results: { value: Uint8Array }[] }>(client, KRPC.Response);
|
||||
const status = decodeMessage<{ version: string }>(KRPC.Status, callResp.results[0]!.value);
|
||||
expect(status.version).toBe('0.5.0');
|
||||
|
||||
client.destroy();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"extends": "../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"outDir": "./dist",
|
||||
"rootDir": "./src",
|
||||
"types": ["node"]
|
||||
},
|
||||
"include": ["src/**/*"]
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
import { defineConfig } from 'vitest/config';
|
||||
|
||||
export default defineConfig({
|
||||
test: {
|
||||
include: ['tests/**/*.test.ts'],
|
||||
environment: 'node',
|
||||
},
|
||||
});
|
||||
Reference in New Issue
Block a user