Add frame processor support for audio streams (node version)

.changeset/soft-paws-sink.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': patch
+---
+
+Add frame processor support for audio streams

packages/livekit-rtc/src/audio_frame.ts

@@ -11,6 +11,8 @@ export class AudioFrame {
   channels: number;
   samplesPerChannel: number;
 
+  private _userdata: Record<string, any> = {};
+
   // note: if converting from Uint8Array to Int16Array, *do not* use buffer.slice!
   // it is marked unstable by Node and can cause undefined behaviour, such as massive chunks of
   // noise being added to the end.
@@ -51,6 +53,11 @@ export class AudioFrame {
       samplesPerChannel: this.samplesPerChannel,
     });
   }
+
+  /** Returns the user data associated with the audio frame. */
+  get userdata() {
+    return this._userdata;
+  }
 }
 
 /**

packages/livekit-rtc/src/audio_stream.ts

@@ -5,12 +5,14 @@ import type { UnderlyingSource } from 'node:stream/web';
 import { AudioFrame } from './audio_frame.js';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
+import { FrameProcessor } from './frame_processor.js';
+import { log } from './log.js';
 import type { NewAudioStreamResponse } from './proto/audio_frame_pb.js';
 import { AudioStreamType, NewAudioStreamRequest } from './proto/audio_frame_pb.js';
 import type { Track } from './track.js';
 
 export interface AudioStreamOptions {
-  noiseCancellation?: NoiseCancellationOptions;
+  noiseCancellation?: NoiseCancellationOptions | FrameProcessor<AudioFrame>;
   sampleRate?: number;
   numChannels?: number;
   frameSizeMs?: number;
@@ -26,7 +28,8 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
   private ffiHandle: FfiHandle;
   private sampleRate: number;
   private numChannels: number;
-  private ncOptions?: NoiseCancellationOptions;
+  private legacyNcOptions?: NoiseCancellationOptions;
+  private frameProcessor?: FrameProcessor<AudioFrame>;
   private frameSizeMs?: number;
 
   constructor(
@@ -37,7 +40,11 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
     if (sampleRateOrOptions !== undefined && typeof sampleRateOrOptions !== 'number') {
       this.sampleRate = sampleRateOrOptions.sampleRate ?? 48000;
       this.numChannels = sampleRateOrOptions.numChannels ?? 1;
-      this.ncOptions = sampleRateOrOptions.noiseCancellation;
+      if (sampleRateOrOptions.noiseCancellation instanceof FrameProcessor) {
+        this.frameProcessor = sampleRateOrOptions.noiseCancellation;
+      } else {
+        this.legacyNcOptions = sampleRateOrOptions.noiseCancellation;
+      }
       this.frameSizeMs = sampleRateOrOptions.frameSizeMs;
     } else {
       this.sampleRate = (sampleRateOrOptions as number) ?? 48000;
@@ -50,10 +57,10 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
       sampleRate: this.sampleRate,
       numChannels: this.numChannels,
       frameSizeMs: this.frameSizeMs,
-      ...(this.ncOptions
+      ...(this.legacyNcOptions
         ? {
-            audioFilterModuleId: this.ncOptions.moduleId,
-            audioFilterOptions: JSON.stringify(this.ncOptions.options),
+            audioFilterModuleId: this.legacyNcOptions.moduleId,
+            audioFilterOptions: JSON.stringify(this.legacyNcOptions.options),
           }
         : {}),
     });
@@ -85,7 +92,14 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
     const streamEvent = ev.message.value.message;
     switch (streamEvent.case) {
       case 'frameReceived':
-        const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
+        let frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
+        if (this.frameProcessor && this.frameProcessor.isEnabled()) {
+          try {
+            frame = this.frameProcessor.process(frame);
+          } catch (err: unknown) {
+            log.warn(`Frame processing failed, passing through original frame: ${err}`);
+          }
+        }
         this.controller.enqueue(frame);
         break;
       case 'eos':

packages/livekit-rtc/src/frame_processor.ts

@@ -0,0 +1,27 @@
+// SPDX-FileCopyrightText: 2025 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { AudioFrame } from './audio_frame.js';
+import { VideoFrame } from './video_frame.js';
+
+export type FrameProcessorStreamInfo = {
+  roomName: string;
+  participantIdentity: string;
+  publicationSid: string;
+};
+
+export type FrameProcessorCredentials = {
+  token: string;
+  url: string;
+};
+
+export abstract class FrameProcessor<Frame extends VideoFrame | AudioFrame> {
+  abstract isEnabled(): boolean;
+  abstract setEnabled(enabled: boolean): void;
+
+  abstract onStreamInfoUpdated(info: FrameProcessorStreamInfo): void;
+  abstract onCredentialsUpdated(credentials: FrameProcessorCredentials): void;
+
+  abstract process(frame: Frame): Frame;
+  abstract close(): void;
+}

packages/livekit-rtc/src/index.ts

@@ -50,3 +50,8 @@ export type { ChatMessage } from './types.js';
 export { VideoFrame } from './video_frame.js';
 export { VideoSource } from './video_source.js';
 export { VideoStream, type VideoFrameEvent } from './video_stream.js';
+export {
+  FrameProcessor,
+  type FrameProcessorStreamInfo,
+  type FrameProcessorCredentials,
+} from './frame_processor.js';