@primafuture/remote-async-call

Typed RPC + Pub/Sub over framed WebSockets with AbortSignal cancellation, streaming, broker forwarding, and lifecycle cleanup.

Installation

npm install @primafuture/remote-async-call

Browser build without a bundler:

<script src="https://www.unpkg.com/@primafuture/remote-async-call/dist/rac.global.js"></script>

The UMD bundle is exposed as rac.

Quick Start

import WebSocket from 'ws';
import {
	FrameCodec,
	RACProxy,
	WebsocketCommunicationConnection,
	type ProcedureSpec,
} from '@primafuture/remote-async-call';

type Procedures = {
	ping: ProcedureSpec<[], 'pong'>;
};

type Events = {
	totalChanged: number;
};

const socket = new WebSocket('ws://localhost:8080');
const racProxy = new RACProxy<Procedures, Events>(
	new WebsocketCommunicationConnection(socket, {
		frameCodec: new FrameCodec(),
	}),
);

await racProxy.register('ping', async () => {
	return 'pong';
});

racProxy.onRemoteProcedureRegistered('ping', async () => {
	const result = await racProxy.call('ping', []);
	console.log(result);
}, { once: true });

WebsocketCommunicationConnection always uses the binary framing protocol. Plain JSON text messages are rejected as protocol errors.

Typed RPC

import {
	PassThroughCommunicationConnection,
	RACProxy,
	type ProcedureSpec,
} from '@primafuture/remote-async-call';

type Procedures = {
	sum: ProcedureSpec<[number, number], number>;
};

type Events = {};

const racProxy = new RACProxy<Procedures, Events>(
	new PassThroughCommunicationConnection(),
);

await racProxy.register('sum', ([a, b]) => {
	return a + b;
});

const result = await racProxy.call('sum', [2, 3]);
console.log(result); // 5

await racProxy.unregister('sum');

ProcedureSpec<Input, Output, Yield, Next> describes the wire input, return value, optional generator yield value, and optional generator next value.

AbortSignal Cancellation

const controller = new AbortController();

const resultPromise = racProxy.call('slowProcedure', { id: 1 }, {
	signal: controller.signal,
});

controller.abort();
await resultPromise; // rejects with AbortError

Registered endpoints receive the same signal:

await racProxy.register('slowProcedure', async (input, context) => {
	context.signal.throwIfAborted?.();
	return input.id;
});

If a remote call is aborted, RAC removes local result listeners, sends a remote cancel event when possible, and aborts the endpoint ProcedureContext.signal.

Bidirectional Streaming RPC

type Procedures = {
	sequence: ProcedureSpec<{ base: number }, number, string, boolean>;
};

await racProxy.register('sequence', async function *(input) {
	const shouldDouble = yield `base:${input.base}`;
	return shouldDouble ? input.base * 2 : input.base;
});

const iterator = await racProxy.call('sequence', { base: 5 });

console.log(await iterator.next());
console.log(await iterator.next(true));

If the caller closes the iterator with return() or aborts the call signal, the remote endpoint receives cancellation and generator cleanup runs.

Pub/Sub

type Events = {
	totalChanged: number;
};

const unsubscribe = await racProxy.subscribe('totalChanged', (value) => {
	console.log('new total', value);
});

await racProxy.publish('totalChanged', 42);

unsubscribe();

Async iterator subscriptions clean themselves up:

const controller = new AbortController();

for await (const value of racProxy.subscribeIterator('totalChanged', {
	signal: controller.signal,
})) {
	console.log(value);
	break;
}

publish() awaits local handlers before it resolves. subscribe() returns an Unsubscribe function.

Lifecycle Cleanup

DisposeScope groups lifecycle resources and disposes them in reverse registration order.

import { DisposeScope } from '@primafuture/remote-async-call';

const scope = new DisposeScope({
	onError: (error, context) => {
		console.error(`cleanup failed during ${context.phase}`, error);
	},
});

const unsubscribe = await racProxy.subscribe('totalChanged', (value) => {
	console.log(value);
});

scope.add(unsubscribe);
scope.interval(() => {
	void racProxy.call('ping', [], { signal: scope.signal }).catch((error) => {
		if (!scope.disposed) {
			console.error('ping failed', error);
		}
	});
}, 30_000);
scope.timeout(() => {
	console.log('still mounted');
}, 5_000);
scope.eventTarget(window, 'online', () => {
	console.log('back online');
});

await scope.disposeAll('component unmounted');

Child scopes inherit the same error reporter:

const connectionScope = scope.child();
const sessionScope = connectionScope.child();

await sessionScope.disposeAll('session changed');
await connectionScope.disposeAll('websocket closed');

Late add() calls on an already disposed scope run immediately. Sync and async errors from late disposers are reported through onError instead of becoming unhandled promise rejections.

Framing And Compression

import {
	CompressionType,
	FrameCodec,
	RACProxy,
	WebsocketCommunicationConnection,
} from '@primafuture/remote-async-call';

const racProxy = new RACProxy(
	new WebsocketCommunicationConnection(socket, {
		frameCodec: new FrameCodec({
			compression: CompressionType.Gzip,
		}),
	}),
);

Compression is explicit. If CompressionStream or DecompressionStream support is missing for the selected compression type, encoding or decoding fails instead of sending misleading compression metadata.

Metadata

Frame metadata can be attached to RPC calls and Pub/Sub publishes.

await racProxy.register('loadUser', async (input, context) => {
	console.log(context.metadata.requestId);
	return { id: input.id };
});

await racProxy.call('loadUser', { id: 'u1' }, {
	metadata: {
		requestId: 'req-1',
	},
});

await racProxy.publish('totalChanged', 42, {
	metadata: {
		requestId: 'req-2',
	},
});

Procedure endpoints receive call metadata through ProcedureContext.metadata. Pub/Sub handlers receive the original PublishOptions as their second argument.

Readiness And Lifecycle Events

const disposeConnected = racProxy.onConnected(() => {
	console.log('transport is connected');
}, { once: true });

const disposeRemote = racProxy.onRemoteSideConnected(() => {
	console.log('remote RAC peer is ready');
});

const disposeReady = racProxy.onRemoteProcedureRegistered('ready', () => {
	console.log('remote ready procedure is available');
}, { once: true });

console.log(racProxy.isProcedureRegisteredRemotely('ready'));
console.log(racProxy.isProcedureRegisteredLocally('ping'));
console.log(racProxy.isTopicSubscribedRemotely('totalChanged'));
console.log(racProxy.isTopicSubscribedLocally('totalChanged'));

disposeConnected();
disposeRemote();
disposeReady();

onConnected() follows the transport. onRemoteSideConnected() fires after the remote RAC peer has exchanged its handshake and re-advertised procedures/subscriptions.

Broker

import { Broker } from '@primafuture/remote-async-call';

const broker = new Broker();
const handle = broker.addParticipant(racProxy);

await handle.remove();

The broker forwards subscriptions and registered procedures between participants. Removing a participant unregisters generated forwarders and internal subscriptions.

API Overview

| Area | Main exports | | --- | --- | | RPC | RACProxy, ProcedureSpec, ProcedureMap, ProcedureContext, CallOptions | | Pub/Sub | EventMap, PublishOptions, PublishEventHandler, Unsubscribe | | Transport | CommunicationConnection, WebsocketCommunicationConnection, PassThroughCommunicationConnection, DummyCommunicationConnection | | Framing | FrameCodec, FrameDecoder, CompressionType, DataFormat, FrameMetadata | | Lifecycle | DisposeScope, Disposer, DisposeScopeOptions | | Broker | Broker, BrokerParticipantHandle | | Low-level utilities | EventEmitter, JsonBufferSerializer, defaultSerializer |

All public imports are named exports. There is no default namespace export.