RPC implementation via FFI

.changeset/itchy-cheetahs-taste.md

@@ -0,0 +1,6 @@
+---
+"example-rpc": minor
+"@livekit/rtc-node": minor
+---
+
+Native RPC support

examples/rpc/.env.example

@@ -0,0 +1,6 @@
+# 1. Copy this file and rename it to .env
+# 2. Update the enviroment variables below.
+
+LIVEKIT_API_KEY=mykey
+LIVEKIT_API_SECRET=mysecret
+LIVEKIT_URL=wss://myproject.livekit.cloud

examples/rpc/README.md

@@ -0,0 +1,39 @@
+# RPC Example
+
+This example demonstrates how to use RPC between two participants with LiveKit.
+
+The example includes two scenarios:
+1. A simple greeting exchange.
+2. A contrived function-calling service with JSON data payloads and multiple method types.
+
+## Prerequisites
+
+Before running this example, make sure you have:
+
+1. Node.js installed on your machine.
+2. A LiveKit server running (either locally or remotely).
+3. LiveKit API key and secret.
+
+## Setup
+
+1. Install dependencies:
+   ```
+   pnpm install
+   ```
+
+2. Create a `.env.local` file in the example directory with your LiveKit credentials:
+   ```
+   LIVEKIT_API_KEY=your_api_key
+   LIVEKIT_API_SECRET=your_api_secret
+   LIVEKIT_URL=your_livekit_url
+   ```
+
+## Running the Example
+
+To run the example, use the following command:
+
+```
+pnpm run start
+```
+
+The example will log to your terminal.

examples/rpc/index.ts

@@ -0,0 +1,176 @@
+import { Room, RpcError } from '@livekit/rtc-node';
+import type { RemoteParticipant } from '@livekit/rtc-node';
+import { randomBytes } from 'crypto';
+import { config } from 'dotenv';
+import { AccessToken } from 'livekit-server-sdk';
+
+config({ path: '.env.local', override: false });
+const LIVEKIT_API_KEY = process.env.LIVEKIT_API_KEY;
+const LIVEKIT_API_SECRET = process.env.LIVEKIT_API_SECRET;
+const LIVEKIT_URL = process.env.LIVEKIT_URL;
+if (!LIVEKIT_API_KEY || !LIVEKIT_API_SECRET || !LIVEKIT_URL) {
+  throw new Error('Missing required environment variables. Please check your .env.local file.');
+}
+
+async function main() {
+  const roomName = `rpc-test-${randomBytes(4).toString('hex')}`;
+
+  console.log(`Connecting participants to room: ${roomName}`);
+
+  const [requestersRoom, greetersRoom, mathGeniusRoom] = await Promise.all([
+    connectParticipant('requester', roomName),
+    connectParticipant('greeter', roomName),
+    connectParticipant('math-genius', roomName),
+  ]);
+
+  // Register all methods for the receiving participant
+  await registerReceiverMethods(greetersRoom, mathGeniusRoom);
+
+  try {
+    console.log('\n\nRunning greeting example...');
+    await Promise.all([performGreeting(requestersRoom)]);
+  } catch (error) {
+    console.error('Error:', error);
+  }
+
+  try {
+    console.log('\n\nRunning math example...');
+    await Promise.all([
+      performSquareRoot(requestersRoom)
+        .then(() => new Promise<void>((resolve) => setTimeout(resolve, 2000)))
+        .then(() => performQuantumHypergeometricSeries(requestersRoom)),
+    ]);
+  } catch (error) {
+    console.error('Error:', error);
+  }
+
+  console.log('\n\nParticipants done, disconnecting...');
+  await requestersRoom.disconnect();
+  await greetersRoom.disconnect();
+  await mathGeniusRoom.disconnect();
+
+  console.log('Participants disconnected. Example completed.');
+
+  process.exit(0);
+}
+
+const registerReceiverMethods = async (greetersRoom: Room, mathGeniusRoom: Room): Promise<void> => {
+  await greetersRoom.localParticipant?.registerRpcMethod(
+    'arrival',
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    async (requestId: string, sender: RemoteParticipant, payload: string, responseTimeoutMs: number) => {
+      console.log(`[Greeter] Oh ${sender.identity} arrived and said "${payload}"`);
+      await new Promise((resolve) => setTimeout(resolve, 2000));
+      return "Welcome and have a wonderful day!";
+    },
+  );
+
+  await mathGeniusRoom.localParticipant?.registerRpcMethod(
+    'square-root',
+    async (requestId: string, sender: RemoteParticipant, payload: string, responseTimeoutMs: number) => {
+      const jsonData = JSON.parse(payload);
+      const number = jsonData.number;
+      console.log(
+        `[Math Genius] I guess ${sender.identity} wants the square root of ${number}. I've only got ${responseTimeoutMs / 1000} seconds to respond but I think I can pull it off.`,
+      );
+
+      console.log(`[Math Genius] *doing math*…`);
+      await new Promise((resolve) => setTimeout(resolve, 2000));
+
+      const result = Math.sqrt(number);
+      console.log(`[Math Genius] Aha! It's ${result}`);
+      return JSON.stringify({ result });
+    },
+  );
+};
+
+const performGreeting = async (room: Room): Promise<void> => {
+  console.log('[Requester] Letting the greeter know that I\'ve arrived');
+  try {
+    const response = await room.localParticipant!.performRpcRequest('greeter', 'arrival', 'Hello');
+    console.log(`[Requester] That's nice, the greeter said: "${response}"`);
+  } catch (error) {
+    console.error('[Requester] RPC call failed:', error);
+    throw error;
+  }
+};
+
+const performSquareRoot = async (room: Room): Promise<void> => {
+  console.log("[Requester] What's the square root of 16?");
+  try {
+    const response = await room.localParticipant!.performRpcRequest('math-genius', 'square-root', JSON.stringify({ number: 16 }));
+    const parsedResponse = JSON.parse(response);
+    console.log(`[Requester] Nice, the answer was ${parsedResponse.result}`);
+  } catch (error) {
+    console.error('[Requester] RPC call failed:', error);
+    throw error;
+  }
+};
+
+const performQuantumHypergeometricSeries = async (room: Room): Promise<void> => {
+  console.log("[Requester] What's the quantum hypergeometric series of 42?");
+  try {
+    const response = await room.localParticipant!.performRpcRequest(
+      'math-genius',
+      'quantum-hypergeometric-series',
+      JSON.stringify({ number: 42 })
+    );
+    const parsedResponse = JSON.parse(response);
+    console.log(`[Requester] genius says ${parsedResponse.result}!`);
+  } catch (error) {
+    if (error instanceof RpcError) {
+      if (error.code === RpcError.ErrorCode.UNSUPPORTED_METHOD) {
+        console.log(`[Requester] Aww looks like the genius doesn't know that one.`);
+        return;
+      }
+    }
+
+    console.error('[Requester] Unexpected error:', error);
+    throw error;
+  }
+};
+
+const createToken = (identity: string, roomName: string) => {
+  const token = new AccessToken(LIVEKIT_API_KEY, LIVEKIT_API_SECRET, {
+    identity,
+  });
+  token.addGrant({
+    room: roomName,
+    roomJoin: true,
+    canPublish: true,
+    canSubscribe: true,
+  });
+  return token.toJwt();
+};
+
+const connectParticipant = async (identity: string, roomName: string): Promise<Room> => {
+  const room = new Room();
+  const token = await createToken(identity, roomName);
+
+  room.on('disconnected', () => {
+    console.log(`[${identity}] Disconnected from room`);
+  });
+
+  await room.connect(LIVEKIT_URL, token);
+
+  await Promise.race([
+    new Promise<void>((resolve) => {
+      if (room.remoteParticipants.size > 0) {
+        resolve();
+      } else {
+        const onParticipantConnected = () => {
+          room.off('participantConnected', onParticipantConnected);
+          resolve();
+        };
+        room.on('participantConnected', onParticipantConnected);
+      }
+    }),
+    new Promise<void>((_, reject) => {
+      setTimeout(() => reject(new Error('Timed out waiting for participants')), 5000);
+    }),
+  ]);
+
+  return room;
+};
+
+main();

examples/rpc/package.json

@@ -0,0 +1,23 @@
+{
+  "name": "example-rpc",
+  "author": "LiveKit",
+  "private": "true",
+  "description": "Example of using RPC in LiveKit",
+  "type": "module",
+  "main": "index.ts",
+  "scripts": {
+    "lint": "eslint -f unix \"**/*.ts\"",
+    "start": "tsx index.ts"
+  },
+  "keywords": [],
+  "license": "Apache-2.0",
+  "dependencies": {
+    "@livekit/rtc-node": "workspace:*",
+    "dotenv": "^16.4.5",
+    "livekit-server-sdk": "workspace:*"
+  },
+  "devDependencies": {
+    "@types/node": "^20.10.4",
+    "tsx": "^4.7.1"
+  }
+}

packages/livekit-rtc/README.md

@@ -76,10 +76,57 @@ await room.localParticipant.publishTrack(track, options);
 await source.captureFrame(new AudioFrame(buffer, 16000, 1, buffer.byteLength / 2));
 ```
 
+### RPC
+
+RPC you to perform your own predefined method calls from one participant to another. This feature is especially powerful when used with [Agents](https://docs.livekit.io/agents), for instance to forward LLM function calls to your client application.
+
+#### Registering an RPC method
+
+The participant who will receive a call must first register for the specific method:
+
+```typescript
+room.localParticipant?.registerRpcMethod(
+  'greet',
+  async (requestId: string, sender: RemoteParticipant, payload: string, responseTimeoutMs: number) => {
+    console.log(`Received greeting from ${sender.identity}: ${payload}`);
+    return `Hello, ${sender.identity}!`;
+  }
+);
+```
+
+The request includes a `responseTimeoutMs` field, which informs you how long you have to return a response. If you are unable to respond in time, you can either send an error or let the request time out on the sender's side.
+
+#### Performing an RPC request
+
+The caller may then initiate a request like so:
+
+```typescript
+try {
+  const response = await room.localParticipant!.performRpcRequest(
+    'recipient-identity',
+    'greet',
+    'Hello from RPC!'
+  );
+  console.log('RPC response:', response);
+} catch (error) {
+  console.error('RPC call failed:', error);
+}
+```
+
+You may find it useful to adjust the `responseTimeoutMs` parameter, which allows you to set the amount of time you will wait for a response. We recommend keeping this value as low as possible while still satisfying the constraints of your application.
+
+#### Errors
+
+LiveKit is a dynamic realtime environment and calls can fail for various reasons. 
+
+You may throw errors of the type `RpcError` with a string `message` in an RPC method handler and they will be received on the caller's side with the message intact. Other errors will not be transmitted and will instead arrive to the caller as `1500` ("Application Error"). Other built-in errors are detailed in `RpcError`.
+
 ## Examples
 
-- [`publish-wav`](https://github.com/livekit/node-sdks/tree/main/examples/publish-wav): connect to a room and publish a wave file
+- [`publish-wav`](https://github.com/livekit/node-sdks/tree/main/examples/publish-wav): connect to a room and publish a .wave file
+- [ `rpc`](https://github.com/livekit/node-sdks/tree/main/examples/rpc): simple back-and-forth RPC interaction
+
 
 ## Getting help / Contributing
 
-Please join us on [Slack](https://livekit.io/join-slack) to get help from our devs/community. We welcome your contributions and details can be discussed there.
+Please join us on [Slack](https://livekit.io/join-slack) to get help from our devs & community. We welcome your contributions and details can be discussed there.

packages/livekit-rtc/generate_proto.sh

@@ -24,4 +24,5 @@ protoc \
     $FFI_PROTOCOL/track.proto \
     $FFI_PROTOCOL/video_frame.proto \
     $FFI_PROTOCOL/e2ee.proto \
-    $FFI_PROTOCOL/stats.proto
+    $FFI_PROTOCOL/stats.proto \
+    $FFI_PROTOCOL/rpc.proto

packages/livekit-rtc/src/index.ts

@@ -38,6 +38,7 @@ export {
   TrackPublishOptions,
   ConnectionState,
 } from './proto/room_pb.js';
+export { RpcError } from './rpc.js';
 export { EncryptionType, EncryptionState } from './proto/e2ee_pb.js';
 export { StreamState, TrackKind, TrackSource } from './proto/track_pb.js';
 export { VideoBufferType, VideoRotation, VideoCodec } from './proto/video_frame_pb.js';

packages/livekit-rtc/src/napi/native.d.ts

@@ -3,11 +3,14 @@
 
 /* auto-generated by NAPI-RS */
 
-export function livekitInitialize(callback: (data: Uint8Array) => void, captureLogs: boolean): void;
-export function livekitFfiRequest(data: Uint8Array): Uint8Array;
-export function livekitRetrievePtr(handle: Uint8Array): bigint;
-export function livekitCopyBuffer(ptr: bigint, len: number): Uint8Array;
-export function livekitDispose(): Promise<void>;
+export declare function livekitInitialize(
+  callback: (data: Uint8Array) => void,
+  captureLogs: boolean,
+): void;
+export declare function livekitFfiRequest(data: Uint8Array): Uint8Array;
+export declare function livekitRetrievePtr(handle: Uint8Array): bigint;
+export declare function livekitCopyBuffer(ptr: bigint, len: number): Uint8Array;
+export declare function livekitDispose(): Promise<void>;
 export declare class FfiHandle {
   constructor(handle: bigint);
   dispose(): void;