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/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,9 +51,8 @@ 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();
@@ -300,11 +298,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 +318,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 +333,7 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
         });
       }
       this._signaling.sendSignalingMessage('soac', {
-        id: this._id,
+        id: this._transportId,
         signaling: localDesc,
       });
     }).catch((e) => {
@@ -483,7 +481,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 +507,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 +526,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 +539,7 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
         });
       }
       this._signaling.sendSignalingMessage('soac', {
-        id: this._id,
+        id: this._transportId,
         signaling: localDesc,
       });
     }).catch((e) => {
@@ -674,11 +671,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 +724,20 @@ 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) {
-            this._subscribePromises.get(internalId).resolve(subscription);
-          }
-        } else {
-          // Add track to the existing stream
-          subscribedStream.mediaStream.addTrack(event.track);
+        const subscription = this._subscription.get(internalId);
+        subscription.stream = event.streams[0];
+        if (this._subscribePromises.has(internalId)) {
+          this._subscribePromises.get(internalId).resolve(subscription);
+          this._subscribePromises.delete(internalId);
         }
+        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 +821,7 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
 
   _sendCandidate(candidate) {
     this._signaling.sendSignalingMessage('soac', {
-      id: this._id,
+      id: this._transportId,
       signaling: {
         type: 'candidate',
         candidate: {
@@ -896,16 +877,10 @@ export class ConferencePeerConnectionChannel extends EventDispatcher {
       (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(internalId).stream) {
         this._subscribePromises.get(internalId).resolve(subscription);
+        this._subscribePromises.delete(internalId);
       }
     } else if (this._publishPromises.has(internalId)) {
       const publication = new Publication(sessionId, () => {
@@ -1056,21 +1031,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 +1077,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,16 @@ export class Subscription extends EventDispatcher {
       writable: false,
       value: id,
     });
+    /**
+     * @member {MediaStream | BidirectionalStream} stream
+     * @instance
+     * @memberof Owt.Conference.Subscription
+     */
+    Object.defineProperty(this, 'stream', {
+      configurable: false,
+      writable: false,
+      value: stream,
+    });
     /**
      * @function stop
      * @instance
@@ -328,5 +338,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;
   }
 }