@majkapp/event-bus-rpc

RPC (Remote Procedure Call) over Event Bus - Request/response style invocation over generic event channels.

Features

• 🔄 Request/Response Pattern: Traditional RPC semantics over event bus
• 🎯 Type-Safe Proxies: Create typed proxies that feel like local method calls
• ⚡ Async/Await Support: Full Promise-based API
• 🔌 Generic Implementation: Works with any event bus implementation
• ⏱️ Timeout Handling: Configurable timeouts with proper cleanup
• 🚨 Error Handling: Comprehensive error types and propagation
• 🔀 Bidirectional RPC: Both sides can be clients and servers
• 📦 Version Support: Optional version checking for service compatibility
• 🎭 Service Registry: Dynamic service registration and discovery
• 🔍 Service Discovery: Built-in proxy.ready() for verifying service availability

Installation

npm install @majkapp/event-bus-rpc @majkapp/event-bus-types @majkapp/event-bus-local

Quick Start

Basic Usage

import { RPCBridge } from '@majkapp/event-bus-rpc';
import { EventBus } from '@majkapp/event-bus-local';

const eventBus = new EventBus();
const bridge = new RPCBridge((channelId) => eventBus.channel(channelId));

const mathService = {
  add: (a: number, b: number) => a + b,
  multiply: (a: number, b: number) => a * b
};

await bridge.registerService('math', mathService);

const sum = await bridge.call<number>('math', 'add', [5, 3]);
console.log(sum);

Using Typed Proxies

interface MathService {
  add(a: number, b: number): Promise<number>;
  multiply(a: number, b: number): Promise<number>;
  divide(a: number, b: number): Promise<number>;
}

const mathProxy = await bridge.createProxy<MathService>('math');

const sum = await mathProxy.add(10, 5);
const product = await mathProxy.multiply(4, 7);

API Reference

RPCBridge

The main interface for RPC communication. Combines service registration and client calls.

Constructor

new RPCBridge(
  getChannel: (channelId: string) => QueryableEventChannel,
  options?: RPCBridgeOptions
)

Options:

• requestChannelId?: string - Custom request channel ID (default: 'rpc-request')
• responseChannelId?: string - Custom response channel ID (default: 'rpc-response')
• timeout?: number - Default timeout in milliseconds (default: 100)
• version?: string - RPC protocol version

Methods

registerService(serviceName, definition, options?): Promise

Register a service with methods callable via RPC.

await bridge.registerService('calculator', {
  add: (a: number, b: number) => a + b,
  subtract: (a: number, b: number) => a - b
}, { version: '1.0.0' });

unregisterService(serviceName): boolean

Remove a registered service.

const removed = bridge.unregisterService('calculator');

hasService(serviceName): boolean

Check if a service is registered.

if (bridge.hasService('calculator')) {
  // Service available
}

listServices(): ServiceMetadata[]

Get metadata for all registered services.

const services = bridge.listServices();
services.forEach(s => {
  console.log(`${s.name}: ${s.methods.join(', ')}`);
});

call(service, method, args?, options?): Promise

Make an RPC call.

const result = await bridge.call<number>('math', 'add', [5, 3], {
  timeout: 5000
});

createProxy(serviceName, options?): Promise

Create a typed proxy for a service. All proxies include a special ready() method for service discovery.

interface UserService {
  create(name: string): Promise<User>;
  getById(id: string): Promise<User>;
}

const users = await bridge.createProxy<UserService>('users');
const user = await users.create('Alice');

Proxy Methods:

Every proxy includes a special ready() method for verifying service availability:

const proxy = await bridge.createProxy<MyService>('myService');

// Wait for service to be available (default 1s timeout)
await proxy.ready();

// Custom timeout
await proxy.ready({ timeout: 5000 });

// Throws if service doesn't exist or isn't ready

The ready() method performs a two-phase check:

1. Existence check: Calls hidden __ping__ method to verify service exists
2. Readiness check: If service has its own ready() method, calls it and requires true return value

Services can implement custom readiness logic:

// External database instance
const db = new Database();

await bridge.registerService('database', {
  ready: () => db.isConnected(),  // Must return true for proxy.ready() to resolve
  query: async (sql) => db.execute(sql)
});

const dbProxy = await bridge.createProxy('database');

// Waits for database connection to be established
await dbProxy.ready();

// Now safe to use
const results = await dbProxy.query('SELECT * FROM users');

clear()

Clear all services and pending calls.

bridge.clear();

ServiceRegistry

Lower-level service registration and request handling.

const registry = new ServiceRegistry(requestChannel, responseChannel);

await registry.register('myService', {
  method1: () => 'result'
});

ServiceClient

Lower-level RPC client for making calls.

const client = new ServiceClient(requestChannel, responseChannel, {
  timeout: 5000,
  version: '1.0.0'
});

const result = await client.call('service', 'method', ['arg1', 'arg2']);

Error Handling

The library provides specific error types for different failure scenarios:

RPCTimeoutError

Thrown when a call exceeds the timeout period.

try {
  await bridge.call('slow', 'method', [], { timeout: 1000 });
} catch (error) {
  if (error instanceof RPCTimeoutError) {
    console.log(`Timed out after ${error.timeout}ms`);
  }
}

RPCServiceNotFoundError

Thrown when the requested service doesn't exist.

RPCMethodNotFoundError

Thrown when the service exists but the method doesn't.

catch (error) {
  if (error instanceof RPCMethodNotFoundError) {
    console.log(`Available methods: ${error.availableMethods}`);
  }
}

RPCExecutionError

Thrown when the service method throws an error.

RPCVersionMismatchError

Thrown when client and server versions don't match.

Advanced Patterns

Service Discovery with proxy.ready()

Use proxy.ready() to verify service availability before making calls:

// Wait for service to become available
const service = await bridge.createProxy<MyService>('myService');

try {
  await service.ready({ timeout: 5000 });
  console.log('Service is ready!');
} catch (error) {
  console.error('Service not available');
}

// Now safe to call methods
await service.doWork();

Two-phase readiness check:

// Service with custom initialization
const processorState = { initialized: false, resourcesLoaded: false };

await bridge.registerService('processor', {
  ready: () => {
    return processorState.initialized && processorState.resourcesLoaded;
  },
  process: async (data) => {
    // Complex processing
  }
});

// Client waits for both service existence AND readiness
const processor = await bridge.createProxy('processor');
await processor.ready(); // Waits for __ping__ response AND ready() === true

Without ready() method (simple services):

await bridge.registerService('simple', {
  getValue: () => 'value'
  // No ready() method
});

const simple = await bridge.createProxy('simple');
await simple.ready(); // Just checks __ping__, no readiness verification

Bidirectional RPC

Both sides can register services and make calls:

const serverBridge = new RPCBridge((id) => eventBus.channel(id));
const clientBridge = new RPCBridge((id) => eventBus.channel(id));

await serverBridge.registerService('server', {
  serverMethod: async () => {
    return await serverBridge.call('client', 'clientMethod');
  }
});

await clientBridge.registerService('client', {
  clientMethod: () => 'from client'
});

const result = await clientBridge.call('server', 'serverMethod');

Multiple Isolated Bridges

Create separate RPC channels on the same event bus:

const bridge1 = new RPCBridge((id) => eventBus.channel(id), {
  requestChannelId: 'bridge1-req',
  responseChannelId: 'bridge1-res'
});

const bridge2 = new RPCBridge((id) => eventBus.channel(id), {
  requestChannelId: 'bridge2-req',
  responseChannelId: 'bridge2-res'
});

Concurrent Calls

Make multiple RPC calls in parallel:

const [result1, result2, result3] = await Promise.all([
  bridge.call('service', 'method1'),
  bridge.call('service', 'method2'),
  bridge.call('service', 'method3')
]);

Cascading Service Calls

Services can call other services:

await bridge.registerService('orchestrator', {
  complexOperation: async (data) => {
    const processed = await bridge.call('processor', 'process', [data]);
    const validated = await bridge.call('validator', 'validate', [processed]);
    await bridge.call('logger', 'log', ['Operation complete']);
    return validated;
  }
});

Examples

See the examples/ directory for complete working examples:

• 01-basic-usage.ts - Basic RPC calls and service registration
• 02-service-proxy.ts - Using typed proxies for CRUD operations
• 03-advanced-patterns.ts - Bidirectional RPC, concurrent calls, error handling

Testing

npm test
npm run test:watch
npm run test:coverage

License

MIT