Support update source for subscription (#816)

* Support update source for subscription

* Change codec order and ext in SDP

* Fix sdp and stream arguments issue

* Fix not allowed codec issue

* Fix activeInput undefined issue for selecting audio

Author: starwarfan

source/agent/audio/index.js

@@ -142,7 +142,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             // Notify controller
             const target = {id: activeStreamIds[i], owner, codec};
             ctrlr && rpcClient.remoteCall(ctrlr, 'onAudioActiveness',
-                [belongToRoom, activeInput, target],
+                [belongToRoom, streamId, target],
                 {callback: function(){}});
         });
         engine = selector;

source/agent/conference/conference.js

@@ -1672,6 +1672,10 @@ var Conference = function (rpcClient, selfRpcId) {
       if (update.audio.from && update.audio.from !== audioTrack.from) {
         audioTrack.from = update.audio.from;
         effective = true;
+        // TODO: limit format when updating source
+        if (oldSub.info && oldSub.info.type === 'webrtc') {
+          delete audioTrack.format;
+        }
       }
     }
 
@@ -1683,6 +1687,11 @@ var Conference = function (rpcClient, selfRpcId) {
       if (update.video.from && update.video.from !== videoTrack.from) {
         videoTrack.from = update.video.from;
         effective = true;
+        // TODO: limit format when updating source
+        if (oldSub.info && oldSub.info.type === 'webrtc') {
+          delete videoTrack.format;
+          videoTrack.parameters = {};
+        }
       }
       if (update.video.parameters && (Object.keys(update.video.parameters).length > 0)) {
         videoTrack.parameters = (videoTrack.parameters || {});
@@ -1714,18 +1723,18 @@ var Conference = function (rpcClient, selfRpcId) {
     }
     const err = {};
     if (videoTrack && !validateVideoRequest(oldSub.info.type, videoTrack, err)) {
-      return Promise.reject('Target video stream does NOT satisfy:', (err && err.message));
+      return Promise.reject('Target video stream does NOT satisfy:' + (err && err.message));
     }
     if (audioTrack && !validateAudioRequest(oldSub.info.type, audioTrack, err)) {
-      return Promise.reject('Target audio stream does NOT satisfy', (err && err.message));
+      return Promise.reject('Target audio stream does NOT satisfy' + (err && err.message));
     }
 
     return removeSubscription(subscriptionId)
       .then((result) => {
         return addSubscription(subscriptionId, oldSub.locality, newSubMedia, oldSub.data, oldSub.info);
       }).catch((err) => {
         log.info('Update subscription failed:', err.message ? err.message : err);
-        log.info('And is recovering the previous subscription:', JSON.stringify(old_su));
+        log.info('And is recovering the previous subscription:', JSON.stringify(oldSub));
         return addSubscription(subscriptionId, oldSub.locality, oldSub.media, oldSub.data, oldSub.info)
           .then(() => {
             return Promise.reject('Update subscription failed');

source/agent/conference/stream.js

@@ -346,7 +346,7 @@ class SelectedStream extends ForwardStream {
       activeInput: 'unknown',
       type: 'selecting',
     };
-    super(id, media, info, null);
+    super(id, media, null, info, null);
   }
 }
 

source/agent/conference/subscription.js

@@ -96,7 +96,7 @@ class Subscription {
   // Array of unique StreamId or TrackId of subscription
   froms() {
     return this.media.tracks.map(t => t.from)
-      .filter((t, i, self) => self.findIndex(v => v.id === t.id) === i);
+      .filter((t, i, self) => self.indexOf(t) === i);
   }
 
   // Merge source format and parameters

source/agent/webrtc/sdpInfo.js

@@ -139,7 +139,15 @@ class SdpInfo {
           }
         }
       }
-
+      if (mediaInfo.direction === 'recvonly') {
+        // For subscription
+        // concat(optionals.map((fmt) => fmt.codec.toLowerCase()));
+        mediaInfo.rtp.forEach((rtp) => {
+          if (optionals.findIndex(fmt => isAudioMatchRtp(fmt, rtp)) > -1) {
+            relatedPayloads.add(rtp.payload);
+          }
+        });
+      }
       if (rtpMap.has(selectedPayload)) {
         const selectedRtp = rtpMap.get(selectedPayload);
         rtpMap.forEach(rtp => {
@@ -165,12 +173,15 @@ class SdpInfo {
         mediaInfo.rtcpFb = mediaInfo.rtcpFb.filter(
           (rtcp) => relatedPayloads.has(rtcp.payload));
       }
-      mediaInfo.payloads = mediaInfo.payloads.toString().split(' ')
+      const payloadList = mediaInfo.payloads.toString().split(' ');
+      if (selectedPayload !== -1) {
+        payloadList.unshift(selectedPayload);
+      }
+      mediaInfo.payloads = payloadList
         .filter((p) => relatedPayloads.has(parseInt(p)))
         .filter((v, index, self) => self.indexOf(v) === index)
         .join(' ');
     }
-
     return finalFmt;
   }
 
@@ -202,6 +213,13 @@ class SdpInfo {
         .forEach((p, index) => {
           payloadOrder.set(parseInt(p), index);
         });
+      if (mediaInfo.direction === 'recvonly') {
+        // For subscription
+        // concat(optionals.map((fmt) => fmt.codec.toLowerCase()));
+        optionals.forEach((fmt) => {
+          reservedCodecs.push(fmt.codec.toLowerCase());
+        });
+      }
 
       for (let i = 0; i < mediaInfo.rtp.length; i++) {
         rtp = mediaInfo.rtp[i];
@@ -245,7 +263,11 @@ class SdpInfo {
         mediaInfo.rtcpFb = mediaInfo.rtcpFb.filter(
           (rtcp) => relatedPayloads.has(rtcp.payload));
       }
-      mediaInfo.payloads = mediaInfo.payloads.toString().split(' ')
+      const payloadList = mediaInfo.payloads.toString().split(' ');
+      if (selectedPayload !== -1) {
+        payloadList.unshift(selectedPayload);
+      }
+      mediaInfo.payloads = payloadList
         .filter((p) => relatedPayloads.has(parseInt(p)))
         .filter((v, index, self) => self.indexOf(v) === index)
         .join(' ');
@@ -530,6 +552,23 @@ class SdpInfo {
         mediaInfo.direction = 'recvonly';
       }
 
+      if (mediaInfo.ext && Array.isArray(mediaInfo.ext)) {
+        const extMappings = [
+          'urn:ietf:params:rtp-hdrext:ssrc-audio-level',
+          'http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01',
+          'urn:ietf:params:rtp-hdrext:sdes:mid',
+          'urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id',
+          'urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id',
+          'urn:ietf:params:rtp-hdrext:toffset',
+          'http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time',
+          // 'urn:3gpp:video-orientation',
+          // 'http://www.webrtc.org/experiments/rtp-hdrext/playout-delay',
+        ];
+        mediaInfo.ext = mediaInfo.ext.filter((e) => {
+          return extMappings.includes(e.uri);
+        });
+      }
+
       if (mediaInfo.rids && Array.isArray(mediaInfo.rids)) {
         // Reverse rids direction
         mediaInfo.rids.forEach(r => {