Dominant Speaker Event (#603)

Implementation of Dominant Speaker Identification for Multipoint Videoconferencing by Ilana Volfin and Israel Cohen. This implementation uses the RTP Audio Level extension from RFC-6464 for the input signal

lib/ActiveSpeakerObserver.d.ts

@@ -0,0 +1,28 @@
+import { EnhancedEventEmitter } from './EnhancedEventEmitter';
+import { RtpObserver } from './RtpObserver';
+import { Producer } from './Producer';
+export interface ActiveSpeakerObserverOptions {
+    interval?: number;
+    /**
+     * Custom application data.
+     */
+    appData?: any;
+}
+export interface ActiveSpeakerObserverActivity {
+    /**
+     * The producer instance.
+     */
+    producer: Producer;
+}
+export declare class ActiveSpeakerObserver extends RtpObserver {
+    /**
+     * @private
+     */
+    constructor(params: any);
+    /**
+     * Observer.
+     */
+    get observer(): EnhancedEventEmitter;
+    private _handleWorkerNotifications;
+}
+//# sourceMappingURL=ActiveSpeakerObserver.d.ts.map

lib/ActiveSpeakerObserver.js

@@ -0,0 +1,40 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });
+const Logger_1 = require("./Logger");
+const RtpObserver_1 = require("./RtpObserver");
+const logger = new Logger_1.Logger('ActiveSpeakerObserver');
+class ActiveSpeakerObserver extends RtpObserver_1.RtpObserver {
+    /**
+     * @private
+     */
+    constructor(params) {
+        super(params);
+        this._handleWorkerNotifications();
+    }
+    /**
+     * Observer.
+     */
+    get observer() {
+        return this._observer;
+    }
+    _handleWorkerNotifications() {
+        this._channel.on(this._internal.rtpObserverId, (event, data) => {
+            switch (event) {
+                case 'dominantspeaker':
+                    {
+                        const dominantSpeaker = {
+                            producer: this._getProducerById(data.producerId)
+                        };
+                        this.safeEmit('dominantspeaker', dominantSpeaker);
+                        this._observer.safeEmit('dominantspeaker', dominantSpeaker);
+                        break;
+                    }
+                default:
+                    {
+                        logger.error('ignoring unknown event "%s"', event);
+                    }
+            }
+        });
+    }
+}
+exports.ActiveSpeakerObserver = ActiveSpeakerObserver;

lib/Router.d.ts

@@ -10,6 +10,7 @@ import { Producer } from './Producer';
 import { Consumer } from './Consumer';
 import { DataProducer } from './DataProducer';
 import { DataConsumer } from './DataConsumer';
+import { ActiveSpeakerObserver, ActiveSpeakerObserverOptions } from './ActiveSpeakerObserver';
 import { AudioLevelObserver, AudioLevelObserverOptions } from './AudioLevelObserver';
 import { RtpCapabilities, RtpCodecCapability } from './RtpParameters';
 import { NumSctpStreams } from './SctpParameters';
@@ -167,6 +168,10 @@ export declare class Router extends EnhancedEventEmitter {
      * Pipes the given Producer or DataProducer into another Router in same host.
      */
     pipeToRouter({ producerId, dataProducerId, router, listenIp, enableSctp, numSctpStreams, enableRtx, enableSrtp }: PipeToRouterOptions): Promise<PipeToRouterResult>;
+    /**
+     * Create an ActiveSpeakerObserver
+     */
+    createActiveSpeakerObserver({ interval, appData }?: ActiveSpeakerObserverOptions): Promise<ActiveSpeakerObserver>;
     /**
      * Create an AudioLevelObserver.
      */

lib/Router.js

@@ -10,6 +10,7 @@ const WebRtcTransport_1 = require("./WebRtcTransport");
 const PlainTransport_1 = require("./PlainTransport");
 const PipeTransport_1 = require("./PipeTransport");
 const DirectTransport_1 = require("./DirectTransport");
+const ActiveSpeakerObserver_1 = require("./ActiveSpeakerObserver");
 const AudioLevelObserver_1 = require("./AudioLevelObserver");
 const logger = new Logger_1.Logger('Router');
 class Router extends EnhancedEventEmitter_1.EnhancedEventEmitter {
@@ -516,6 +517,31 @@ class Router extends EnhancedEventEmitter_1.EnhancedEventEmitter {
             throw new Error('internal error');
         }
     }
+    /**
+     * Create an ActiveSpeakerObserver
+     */
+    async createActiveSpeakerObserver({ interval = 300, appData = {} } = {}) {
+        logger.debug('createActiveSpeakerObserver()');
+        if (appData && typeof appData !== 'object')
+            throw new TypeError('if given, appData must be an object');
+        const internal = { ...this._internal, rtpObserverId: uuid_1.v4() };
+        const reqData = { interval };
+        await this._channel.request('router.createActiveSpeakerObserver', internal, reqData);
+        const activeSpeakerObserver = new ActiveSpeakerObserver_1.ActiveSpeakerObserver({
+            internal,
+            channel: this._channel,
+            payloadChannel: this._payloadChannel,
+            appData,
+            getProducerById: (producerId) => (this._producers.get(producerId))
+        });
+        this._rtpObservers.set(activeSpeakerObserver.id, activeSpeakerObserver);
+        activeSpeakerObserver.on('@close', () => {
+            this._rtpObservers.delete(activeSpeakerObserver.id);
+        });
+        // Emit observer event.
+        this._observer.safeEmit('newrtpobserver', activeSpeakerObserver);
+        return activeSpeakerObserver;
+    }
     /**
      * Create an AudioLevelObserver.
      */

lib/types.d.ts

@@ -10,6 +10,7 @@ export * from './Consumer';
 export * from './DataProducer';
 export * from './DataConsumer';
 export * from './RtpObserver';
+export * from './ActiveSpeakerObserver';
 export * from './AudioLevelObserver';
 export * from './RtpParameters';
 export * from './SctpParameters';

lib/types.js

@@ -15,5 +15,6 @@ __export(require("./Consumer"));
 __export(require("./DataProducer"));
 __export(require("./DataConsumer"));
 __export(require("./RtpObserver"));
+__export(require("./ActiveSpeakerObserver"));
 __export(require("./AudioLevelObserver"));
 __export(require("./errors"));

rust/src/lib.rs

@@ -68,6 +68,12 @@ pub mod audio_level_observer {
     pub use crate::router::audio_level_observer::*;
 }
 
+pub mod active_speaker_observer {
+    //! An active speaker observer monitors the speaking activity of the selected audio producers.
+
+    pub use crate::router::active_speaker_observer::*;
+}
+
 pub mod consumer {
     //! A consumer represents an audio or video source being forwarded from a mediasoup router to an
     //! endpoint. It's created on top of a transport that defines how the media packets are carried.

rust/src/messages.rs

@@ -1,3 +1,4 @@
+use crate::active_speaker_observer::ActiveSpeakerObserverOptions;
 use crate::audio_level_observer::AudioLevelObserverOptions;
 use crate::consumer::{
     ConsumerDump, ConsumerId, ConsumerLayers, ConsumerScore, ConsumerStats, ConsumerTraceEventType,
@@ -414,6 +415,30 @@ request_response!(
     },
 );
 
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct RouterCreateActiveSpeakerObserverData {
+    pub(crate) interval: u16,
+}
+
+impl RouterCreateActiveSpeakerObserverData {
+    pub(crate) fn from_options(
+        active_speaker_observer_options: &ActiveSpeakerObserverOptions,
+    ) -> Self {
+        Self {
+            interval: active_speaker_observer_options.interval,
+        }
+    }
+}
+
+request_response!(
+    "router.createActiveSpeakerObserver",
+    RouterCreateActiveSpeakerObserverRequest {
+        internal: RtpObserverInternal,
+        data: RouterCreateActiveSpeakerObserverData,
+    },
+);
+
 request_response!(
     "transport.close",
     TransportCloseRequest {