Add volume and document to active audio event (#1021)

Author: starwarfan

doc/design/active-audio-selecting.md

@@ -0,0 +1,29 @@
+| OWT Active Audio Stream Introduction |
+
+
+# Overview
+
+OWT conference provide functionality to select K (K = 3 by default) most active audio streams in the conference.
+
+# How to enable
+The active audio selecting feature can be enabled by setting `selectActiveAudio` to true in room configuration. When it's enabled, an audio node will be used for selecting audio streams for each room, and K audio-only streams will be produced as most active audio streams in the conference.
+
+# Stream
+```
+stream.info = {
+  type: 'selecting',
+  owner: 'unknown' | string(ParticipantId),
+  activeInput: 'unknown' | string(StreamId),
+}
+```
+
+# Events
+Stream notification for input change of active audio stream.
+```
+{
+  id: activeAudioId,
+  status: 'update',
+  data: {field: 'activeInput', value: {id: originInput, volume: 'major' | 'minor'}}
+}
+```
+

source/agent/audio/activeAudioSelector.js

@@ -31,6 +31,7 @@ class ActiveAudioSelector extends EventEmitter {
             this.ranker.addOutput(this.topK[i]);
         }
         this.currentRank = Array(k).fill('');
+        this.currentVolume = Array(k).fill('');
         log.debug('Init with K:', k);
     }
 
@@ -55,6 +56,15 @@ class ActiveAudioSelector extends EventEmitter {
                     log.warn('Unknown stream ID', streamId);
                 }
                 pendingAddIndexes.push(i);
+            } else if (this.currentVolume[i] !== changes[i][1]) {
+                let streamId = this.currentRank[i];
+                if (this.inputs.has(streamId)) {
+                    let { owner, codec } = this.inputs.get(streamId);
+                    const volume = changes[i][1];
+                    log.debug('Volume change:', streamId, volume);
+                    this.emit('source-change',
+                        i, owner, streamId, codec, volume);
+                }
             }
         }
         for (let i of pendingAddIndexes) {
@@ -63,12 +73,14 @@ class ActiveAudioSelector extends EventEmitter {
             if (this.inputs.has(streamId)) {
                 let { owner, codec } = this.inputs.get(streamId);
                 log.debug('Active input:', owner, streamId, codec, i);
-                this.emit('source-change', i, owner, streamId, codec);
+                const volume = changes[i][1];
+                this.emit('source-change', i, owner, streamId, codec, volume);
             } else if (streamId !== '') {
                 log.warn('Unknown stream ID', streamId);
             }
         }
         this.currentRank = changes.map(pair => pair[0]);
+        this.currentVolume = changes.map(pair => pair[1]);
     }
 
     addInput(owner, streamId, codec, conn) {

source/agent/audio/index.js

@@ -138,9 +138,9 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             return;
         }
         const selector = new ActiveAudioSelector(activeStreamIds.length);
-        selector.on('source-change', (i, owner, streamId, codec) => {
+        selector.on('source-change', (i, owner, streamId, codec, volume) => {
             // Notify controller
-            const target = {id: activeStreamIds[i], owner, codec};
+            const target = {id: activeStreamIds[i], owner, codec, volume};
             ctrlr && rpcClient.remoteCall(ctrlr, 'onAudioActiveness',
                 [belongToRoom, streamId, target],
                 {callback: function(){}});

source/agent/conference/conference.js

@@ -1927,12 +1927,18 @@ var Conference = function (rpcClient, selfRpcId) {
             activeInputStream : trackOwners[activeInputStream];
         const activeAudioId = target.id;
         if (streams[activeAudioId] instanceof SelectedStream) {
-          if (streams[activeAudioId].info.activeInput !== input) {
+          if (streams[activeAudioId].info.activeInput !== input ||
+              streams[activeAudioId].info.volume !== target.volume) {
             streams[activeAudioId].info.activeInput = input;
             streams[activeAudioId].info.activeOwner = target.owner;
-            room_config.notifying.streamChange &&
-                sendMsg('room', 'all', 'stream',
-                    {id: activeAudioId, status: 'update', data: {field: 'activeInput', value: input}});
+            streams[activeAudioId].info.volume = target.volume;
+            //TODO: separate major and activeInput when notifying
+            room_config.notifying.streamChange && sendMsg('room', 'all', 'stream',
+              {
+                id: activeAudioId,
+                status: 'update',
+                data: {field: 'activeInput', value: {id: input, volume: target.volume}}
+              });
           }
         }
         callback('callback', 'ok');

source/agent/conference/stream.js

@@ -40,6 +40,11 @@
  *       stream: string(StreamId),
  *       region: object(Region)
  *     }]
+ *   } | object(SelectInfo):: {
+ *     owner: string(ParticipantId),
+ *     type: 'selecting',
+ *     activeInput: 'unknown' | string(StreamId),
+ *     volume: 'major' | 'minor' | undefined,
  *   }
  * }
  */

source/core/owt_base/selector/AudioRanker.cpp

@@ -160,9 +160,17 @@ void AudioRanker::updateInputInternal(std::string streamId, int level, bool trig
         }
 
         if (inTopK && level != audioProc->iter()->first) {
+            ELOG_DEBUG("Top K internal change: %s %s",
+                audioProc->streamId().c_str(), audioProc->ownerId().c_str());
+            auto oldUp = m_topK.upper_bound(audioProc->iter()->first);
             m_topK.erase(audioProc->iter());
             auto newIt = m_topK.emplace(level, audioProc);
             audioProc->setIter(newIt);
+            auto newUp = m_topK.upper_bound(level);
+            if (oldUp != newUp && triggerChange) {
+                ELOG_DEBUG("Top K internal trigger");
+                triggerRankChange();
+            }
         }
     } else {
         // Previous in others, check if it's still in
@@ -241,14 +249,16 @@ void AudioRanker::triggerRankChange()
             }
         }
 
+        size_t order = 0;
         for (auto& pair : m_topK) {
+            order++;
             auto audioProc = pair.second;
             FrameDestination* output = audioProc->linkedOutput();
             int index = m_outputIndexes[output];
             ELOG_DEBUG("update output index: %d, streamId: %s",
                 index, audioProc->streamId().c_str());
             updates[index].first = audioProc->streamId();
-            updates[index].second = audioProc->ownerId();
+            updates[index].second = (order == m_topK.size() ? "major" : "minor");
         }
 
         bool hasChange = false;
@@ -338,7 +348,7 @@ void AudioRanker::AudioLevelProcessor::onFeedback(const FeedbackMsg& msg)
 
 void AudioRanker::AudioLevelProcessor::deliverOwnerData()
 {
-    ELOG_DEBUG("deliver ownerID as metadata");
+    ELOG_DEBUG("deliver ownerID as metadata %s", m_ownerId.c_str());
     // Pass owner ID metadata to linkedoutput
     MetaData ownerData;
     ownerData.type = META_DATA_OWNER_ID;