Subscribed streams are associated with subscriptions.

When subscribe a stream in conference mode, the subscribed stream no
longer associated with RemoteStream. It allows a RemoteStream to be
subscribed multiple times. It also allows a subscription has audio and
video track from different remote streams.

Since the signaling protocol defined by server side does not provide an
ID for a track, the SDK usually use stream ID + track kind to identify a
track. The stream ID and track ID mentioned in conference mode indicate
the ID assigned by conference sever, they could be different from
MediaStream ID and MediaStreamTrack ID.

Author: jianjunz

docs/mdfiles/changelog.md

@@ -1,5 +1,7 @@
 Change Log
 ==========
+# 5.1
+* When subscribe a stream in conference mode, the subscribed MediaStream or BidirectionalStream is associated with a `Owt.Conference.Subscription` instead of a `Owt.Base.RemoteStream`. The `stream` property of a RemoteStream in conference mode is always undefined, while a new property `stream` is added to `Subscription`. It allows a RemoteStream to be subscribed multiple times, as well as subscribing audio and video tracks from different streams.
 # 5.0
 * Add WebTransport support for conference mode, see [this design doc](../../design/webtransport.md) for detailed information.
 * All publications and subscriptions for the same conference use the same `PeerConnection`.

src/samples/conference/public/scripts/index.js

@@ -94,15 +94,15 @@ const runSocketIOSample = function() {
             }).then((
                 subscription) => {
                     subscirptionLocal = subscription;
-                $(`#${stream.id}`).get(0).srcObject = stream.mediaStream;
+                $(`#${stream.id}`).get(0).srcObject = subscription.stream;
             });
         }
         let $p = createResolutionButtons(stream, subscribeDifferentResolution);
         conference.subscribe(stream)
         .then((subscription)=>{
             subscirptionLocal = subscription;
             let $video = $(`<video controls autoplay id=${stream.id} style="display:block" >this browser does not supported video tag</video>`);
-           $video.get(0).srcObject = stream.mediaStream;
+           $video.get(0).srcObject = subscription.stream;
            $p.append($video);
         }, (err)=>{ console.log('subscribe failed', err);
         });

src/sdk/base/stream.js

@@ -149,7 +149,9 @@ export class LocalStream extends Stream {
 }
 /**
  * @class RemoteStream
- * @classDesc Stream sent from a remote endpoint.
+ * @classDesc Stream sent from a remote endpoint. In conference mode,
+ * RemoteStream's stream is always undefined. Please get MediaStream or
+ * ReadableStream from subscription's stream property.
  * Events:
  *
  * | Event Name      | Argument Type    | Fired when         |

src/sdk/conference/channel.js

@@ -43,7 +43,6 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
     this._pendingCandidates = [];
     this._subscribePromises = new Map(); // internalId => { resolve, reject }
     this._publishPromises = new Map(); // internalId => { resolve, reject }
-    this._subscribedStreams = new Map(); // intenalId => RemoteStream
     this._publications = new Map(); // PublicationId => Publication
     this._subscriptions = new Map(); // SubscriptionId => Subscription
     this._publishTransceivers = new Map(); // internalId => { id, transceivers: [Transceiver] }
@@ -52,15 +51,15 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
     // Timer for PeerConnection disconnected. Will stop connection after timer.
     this._disconnectTimer = null;
     this._ended = false;
-    this._stopped = false;
     // Channel ID assigned by conference
-    this._id = undefined;
+    this._transportId = undefined;
     // Used to create internal ID for publication/subscription
     this._internalCount = 0;
     this._sdpPromise = Promise.resolve();
     this._sdpResolverMap = new Map(); // internalId => {finish, resolve, reject}
     this._sdpResolvers = []; // [{finish, resolve, reject}]
     this._sdpResolveNum = 0;
+    this._remoteMediaStream = new Map(); // Key is subscription ID, value is MediaStream.
   }
 
   /**
@@ -300,11 +299,11 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
       return this._signaling.sendSignalingMessage('publish', {
         media: {tracks: trackOptions},
         attributes: stream.attributes,
-        transport: {id: this._id, type: 'webrtc'},
+        transport: {id: this._transportId, type: 'webrtc'},
       }).catch((e) => {
         // Send SDP even when failed to get Answer.
         this._signaling.sendSignalingMessage('soac', {
-          id: this._id,
+          id: this._transportId,
           signaling: localDesc,
         });
         throw e;
@@ -320,10 +319,10 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
       this._publishTransceivers.get(internalId).id = publicationId;
       this._reverseIdMap.set(publicationId, internalId);
 
-      if (this._id && this._id !== data.transportId) {
+      if (this._transportId && this._transportId !== data.transportId) {
         Logger.warning('Server returns conflict ID: ' + data.transportId);
       }
-      this._id = data.transportId;
+      this._transportId = data.transportId;
 
       // Modify local SDP before sending
       if (options) {
@@ -335,7 +334,7 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
         });
       }
       this._signaling.sendSignalingMessage('soac', {
-        id: this._id,
+        id: this._transportId,
         signaling: localDesc,
       });
     }).catch((e) => {
@@ -483,7 +482,6 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
       offerOptions.offerToReceiveVideo = !!options.video;
     }
     this._subscribeTransceivers.set(internalId, {transceivers});
-    this._subscribedStreams.set(internalId, stream);
 
     let localDesc;
     this._pc.createOffer(offerOptions).then((desc) => {
@@ -510,11 +508,11 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
       });
       return this._signaling.sendSignalingMessage('subscribe', {
         media: {tracks: trackOptions},
-        transport: {id: this._id, type: 'webrtc'},
+        transport: {id: this._transportId, type: 'webrtc'},
       }).catch((e) => {
         // Send SDP even when failed to get Answer.
         this._signaling.sendSignalingMessage('soac', {
-          id: this._id,
+          id: this._transportId,
           signaling: localDesc,
         });
         throw e;
@@ -529,10 +527,10 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
 
       this._subscribeTransceivers.get(internalId).id = subscriptionId;
       this._reverseIdMap.set(subscriptionId, internalId);
-      if (this._id && this._id !== data.transportId) {
+      if (this._transportId && this._transportId !== data.transportId) {
         Logger.warning('Server returns conflict ID: ' + data.transportId);
       }
-      this._id = data.transportId;
+      this._transportId = data.transportId;
 
       // Modify local SDP before sending
       if (options) {
@@ -542,7 +540,7 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
         });
       }
       this._signaling.sendSignalingMessage('soac', {
-        id: this._id,
+        id: this._transportId,
         signaling: localDesc,
       });
     }).catch((e) => {
@@ -674,11 +672,6 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
               new ConferenceError('Failed to subscribe'));
         }
       }
-      // Clear media stream
-      if (this._subscribedStreams.has(internalId)) {
-        this._subscribedStreams.get(internalId).mediaStream = null;
-        this._subscribedStreams.delete(internalId);
-      }
       if (this._sdpResolverMap.has(internalId)) {
         const resolver = this._sdpResolverMap.get(internalId);
         if (!resolver.finish) {
@@ -732,31 +725,24 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
 
   _onRemoteStreamAdded(event) {
     Logger.debug('Remote stream added.');
-    let find = false;
     for (const [internalId, sub] of this._subscribeTransceivers) {
-      const subscriptionId = sub.id;
       if (sub.transceivers.find((t) => t.transceiver === event.transceiver)) {
-        find = true;
-        const subscribedStream = this._subscribedStreams.get(internalId);
-        if (!subscribedStream.mediaStream) {
-          this._subscribedStreams.get(internalId).mediaStream =
-              event.streams[0];
-          // Resolve subscription if ready handler has been called
-          const subscription = this._subscriptions.get(subscriptionId);
-          if (subscription) {
+        if (this._subscriptions.has(sub.id)) {
+          const subscription = this._subscriptions.get(sub.id);
+          subscription.stream = event.streams[0];
+          if (this._subscribePromises.has(internalId)) {
             this._subscribePromises.get(internalId).resolve(subscription);
+            this._subscribePromises.delete(internalId);
           }
         } else {
-          // Add track to the existing stream
-          subscribedStream.mediaStream.addTrack(event.track);
+          this._remoteMediaStream.set(sub.id, event.streams[0]);
         }
+        return;
       }
     }
-    if (!find) {
-      // This is not expected path. However, this is going to happen on Safari
-      // because it does not support setting direction of transceiver.
-      Logger.warning('Received remote stream without subscription.');
-    }
+    // This is not expected path. However, this is going to happen on Safari
+    // because it does not support setting direction of transceiver.
+    Logger.warning('Received remote stream without subscription.');
   }
 
   _onLocalIceCandidate(event) {
@@ -840,7 +826,7 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
 
   _sendCandidate(candidate) {
     this._signaling.sendSignalingMessage('soac', {
-      id: this._id,
+      id: this._transportId,
       signaling: {
         type: 'candidate',
         candidate: {
@@ -887,25 +873,21 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
   }
 
   _readyHandler(sessionId) {
+    Logger.debug('Received ready of '+sessionId);
     const internalId = this._reverseIdMap.get(sessionId);
     if (this._subscribePromises.has(internalId)) {
-      const subscription = new Subscription(sessionId, () => {
+      const mediaStream = this._remoteMediaStream.get(sessionId);
+      const subscription = new Subscription(sessionId, mediaStream, () => {
         this._unsubscribe(internalId);
       }, () => this._getStats(),
       (trackKind) => this._muteOrUnmute(sessionId, true, false, trackKind),
       (trackKind) => this._muteOrUnmute(sessionId, false, false, trackKind),
       (options) => this._applyOptions(sessionId, options));
       this._subscriptions.set(sessionId, subscription);
-      // Fire subscription's ended event when associated stream is ended.
-      this._subscribedStreams.get(internalId).addEventListener('ended', () => {
-        if (this._subscriptions.has(sessionId)) {
-          this._subscriptions.get(sessionId).dispatchEvent(
-              'ended', new OwtEvent('ended'));
-        }
-      });
-      // Resolve subscription if mediaStream is ready
-      if (this._subscribedStreams.get(internalId).mediaStream) {
+      // Resolve subscription if mediaStream is ready.
+      if (this._subscriptions.get(sessionId).stream) {
         this._subscribePromises.get(internalId).resolve(subscription);
+        this._subscribePromises.delete(internalId);
       }
     } else if (this._publishPromises.has(internalId)) {
       const publication = new Publication(sessionId, () => {
@@ -1056,21 +1038,23 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
     return sdp;
   }
 
-  // Handle stream event sent from MCU. Some stream events should be publication
-  // event or subscription event. It will be handled here.
+  // Handle stream event sent from MCU. Some stream update events sent from
+  // server, more specifically audio.status and video.status events should be
+  // publication event or subscription events. They don't change MediaStream's
+  // status. See
+  // https://github.com/open-webrtc-toolkit/owt-server/blob/master/doc/Client-Portal%20Protocol.md#339-participant-is-notified-on-streams-update-in-room
+  // for more information.
   _onStreamEvent(message) {
     const eventTargets = [];
     if (this._publications.has(message.id)) {
       eventTargets.push(this._publications.get(message.id));
     }
-    for (const [internalId, subscribedStream] of this._subscribedStreams) {
-      if (message.id === subscribedStream.id) {
-        const subscriptionId = this._subscribeTransceivers.get(internalId).id;
-        eventTargets.push(this._subscriptions.get(subscriptionId));
-        break;
+    for (const subscription of this._subscriptions) {
+      if (message.id === subscription._audioTrackId ||
+          message.id === subscription._videoTrackId) {
+        eventTargets.push(subscription);
       }
     }
-
     if (!eventTargets.length) {
       return;
     }
@@ -1100,9 +1084,9 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
     // Only check the first one.
     const param = obj[0];
     return !!(
-        param.codecPayloadType || param.dtx || param.active || param.ptime ||
-        param.maxFramerate || param.scaleResolutionDownBy || param.rid ||
-        param.scalabilityMode);
+      param.codecPayloadType || param.dtx || param.active || param.ptime ||
+      param.maxFramerate || param.scaleResolutionDownBy || param.rid ||
+      param.scalabilityMode);
   }
 
   _isOwtEncodingParameters(obj) {

src/sdk/conference/subscription.js

@@ -270,7 +270,7 @@ export class SubscriptionUpdateOptions {
  */
 export class Subscription extends EventDispatcher {
   // eslint-disable-next-line require-jsdoc
-  constructor(id, stop, getStats, mute, unmute, applyOptions) {
+  constructor(id, stream, stop, getStats, mute, unmute, applyOptions) {
     super();
     if (!id) {
       throw new TypeError('ID cannot be null or undefined.');
@@ -285,6 +285,19 @@ export class Subscription extends EventDispatcher {
       writable: false,
       value: id,
     });
+    /**
+     * @member {MediaStream | BidirectionalStream} stream
+     * @instance
+     * @memberof Owt.Conference.Subscription
+     */
+    Object.defineProperty(this, 'stream', {
+      configurable: false,
+      // TODO: It should be a readonly property, but current implementation
+      // creates Subscription after receiving 'ready' from server. At this time,
+      // MediaStream may not be available.
+      writable: true,
+      value: stream,
+    });
     /**
      * @function stop
      * @instance
@@ -328,5 +341,10 @@ export class Subscription extends EventDispatcher {
      * @returns {Promise<undefined, Error>}
      */
     this.applyOptions = applyOptions;
+
+    // Track is not defined in server protocol. So these IDs are equal to their
+    // stream's ID at this time.
+    this._audioTrackId = undefined;
+    this._videoTrackId = undefined;
   }
 }