Update QuicTransport APIs for conference SDK.

It follows the design in docs\webtransport.md.

Author: jianjunz

docs/design/webtransport.md

@@ -24,7 +24,8 @@ JavaScript SDK creates a QuicTransport with a QUIC agent when QUIC agent is enab
 
 ```
 const sendStream = conferenceClient.createSendStream();
-const publication = await conferenceClient.publish(sendStream);
+const localStream = new LocalStream(sendStream, new StreamSourceInfo(undefined, undefined, true));
+const publication = await conferenceClient.publish(localStream);
 sendStream.stream.write(somethingToWrite);
 ```
 

docs/mdfiles/changelog.md

@@ -1,5 +1,8 @@
 Change Log
 ==========
+# 4.4
+* `LocalStream`'s property `mediaStream` is renamed to `stream`. It could also be a `SendStream` or `BidirectionalStream`.
+
 # 4.3
 * The type of `conferenceClient.publish` method's option is changed from `AudioEncodingParameters` and `VideoEncodingParameters` to `RTCRtpSendParameters`.
 * `audio` and `video` of `PublicationSettings` is changed from `AudioPublicationSettings` and `VideoPublicationSettings` to `AudioPublicationSettings` array and `VideoPublicationSettings` array.

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

@@ -63,13 +63,15 @@ function createRandomContentSessionId() {
 }
 
 async function createSendChannel() {
-  bidirectionalStream = await quicChannel.createSendStream();
+  bidirectionalStream = await conference.createSendStream();
+  const localStream=new Owt.Base.LocalStream(bidirectionalStream, new Owt.Base.StreamSourceInfo(undefined, undefined,true));
+  const publication = await conference.publish(localStream);
+  console.log(publication);
   updateConferenceStatus('Created send channel.');
 }
 
 async function windowOnLoad() {
   await joinConference();
-  await createQuicTransport();
   await createSendChannel();
 }
 

src/sdk/base/stream.js

@@ -2,7 +2,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 
-/* global MediaStream */
+/* global MediaStream, SendStream, BidirectionalStream */
 
 'use strict';
 import * as Utils from './utils.js';
@@ -55,21 +55,39 @@ export class Stream extends EventDispatcher {
   // eslint-disable-next-line require-jsdoc
   constructor(stream, sourceInfo, attributes) {
     super();
-    if ((stream && !(stream instanceof MediaStream)) || (typeof sourceInfo !==
-        'object')) {
+    if ((stream && !(stream instanceof MediaStream) &&
+         !(stream instanceof SendStream) &&
+         !(stream instanceof BidirectionalStream)) ||
+        (typeof sourceInfo !== 'object')) {
       throw new TypeError('Invalid stream or sourceInfo.');
     }
-    if (stream && ((stream.getAudioTracks().length > 0 && !sourceInfo.audio) ||
-        stream.getVideoTracks().length > 0 && !sourceInfo.video)) {
+    if (stream && (stream instanceof MediaStream) &&
+        ((stream.getAudioTracks().length > 0 && !sourceInfo.audio) ||
+         stream.getVideoTracks().length > 0 && !sourceInfo.video)) {
       throw new TypeError('Missing audio source info or video source info.');
     }
     /**
      * @member {?MediaStream} mediaStream
      * @instance
      * @memberof Owt.Base.Stream
      * @see {@link https://www.w3.org/TR/mediacapture-streams/#mediastream|MediaStream API of Media Capture and Streams}.
+     * @desc This property is deprecated, please use stream instead.
      */
-    Object.defineProperty(this, 'mediaStream', {
+    if (stream instanceof MediaStream) {
+      Object.defineProperty(this, 'mediaStream', {
+        configurable: false,
+        writable: true,
+        value: stream,
+      });
+    }
+    /**
+     * @member {MediaStream | SendStream | BidirectionalStream | undefined} stream
+     * @instance
+     * @memberof Owt.Base.Stream
+     * @see {@link https://www.w3.org/TR/mediacapture-streams/#mediastream|MediaStream API of Media Capture and Streams}
+     * @see {@link https://wicg.github.io/web-transport/ WebTransport}.
+     */
+    Object.defineProperty(this, 'stream', {
       configurable: false,
       writable: true,
       value: stream,
@@ -112,8 +130,8 @@ export class Stream extends EventDispatcher {
 export class LocalStream extends Stream {
   // eslint-disable-next-line require-jsdoc
   constructor(stream, sourceInfo, attributes) {
-    if (!(stream instanceof MediaStream)) {
-      throw new TypeError('Invalid stream.');
+    if (!stream) {
+      throw new TypeError('Stream cannot be null.');
     }
     super(stream, sourceInfo, attributes);
     /**

src/sdk/conference/client.js

@@ -112,7 +112,7 @@ export const ConferenceClient = function(config, signalingImpl) {
   const participants = new Map(); // Key is participant ID, value is a Participant object.
   const publishChannels = new Map(); // Key is MediaStream's ID, value is pc channel.
   const channels = new Map(); // Key is channel's internal ID, value is channel.
-  let quicTransportChannel = null;
+  let quicTransportChannel;
 
   /**
    * @function onSignalingMessage
@@ -349,6 +349,12 @@ export const ConferenceClient = function(config, signalingImpl) {
    * @param {string} tokenString Token is issued by conference server(nuve).
    */
   this.join = function(tokenString) {
+    if (QuicConnection) {
+      // TODO: Get URL from token.
+      quicTransportChannel = new QuicConnection(
+          'quic-transport://jianjunz-nuc-ubuntu.sh.intel.com:7700/echo',
+          createSignalingForChannel());
+    }
     return new Promise((resolve, reject) => {
       const token = JSON.parse(Base64.decodeBase64(tokenString));
       const isSecured = (token.secure === true);
@@ -414,6 +420,9 @@ export const ConferenceClient = function(config, signalingImpl) {
     if (!(stream instanceof StreamModule.LocalStream)) {
       return Promise.reject(new ConferenceError('Invalid stream.'));
     }
+    if (stream.source.data) {
+      return quicTransportChannel.publish(stream);
+    }
     if (publishChannels.has(stream.mediaStream.id)) {
       return Promise.reject(new ConferenceError(
           'Cannot publish a published stream.'));
@@ -478,18 +487,20 @@ export const ConferenceClient = function(config, signalingImpl) {
     });
   };
 
-    /**
-   * @function createDataStream
+  /**
+   * @function createSendStream
    * @memberOf Owt.Conference.ConferenceClient
    * @instance
-   * @desc Create a bidirectional stream.
-   * @return {Promise<void, Error>} Returned promise will be resolved with a bidirectional stream once stream is created.
+   * @desc Create a outgoing stream. Only available when WebTransport is supported by user's browser.
+   * @return {Promise<SendStream, Error>} Returned promise will be resolved with a SendStream once stream is created.
    */
-  this.createQuicConnection = function() {
-    const quicConnection = new QuicConnection(
-        'quic-transport://jianjunz-nuc-ubuntu.sh.intel.com:7700/echo',
-        createSignalingForChannel());
-    quicTransportChannel = quicConnection;
-    return quicConnection;
-  };
+  if (QuicConnection) {
+    this.createSendStream = async function() {
+      if (!quicTransportChannel) {
+        // Try to create a new one or consider it as closed?
+        throw new ConferenceError('No QUIC connection available.');
+      }
+      return quicTransportChannel.createSendStream();
+    };
+  }
 };

src/sdk/conference/quicconnection.js

@@ -12,6 +12,7 @@ import {
   EventDispatcher,
   MessageEvent,
 } from '../base/event.js';
+import {Publication} from '../base/publication.js';
 import {Subscription} from './subscription.js';
 
 /**
@@ -112,9 +113,16 @@ export class QuicConnection extends EventDispatcher {
     writer.write(new Uint8Array(16));
     const encoder = new TextEncoder();
     writer.write(encoder.encode(token));
+    Logger.info('End of auth.');
   }
 
-  async createSendStream(sessionId) {
+  async createSendStream() {
+    await this._quicTransport.ready;
+    const quicStream = await this._quicTransport.createSendStream();
+    return quicStream;
+  }
+
+  async createSendStream1(sessionId) {
     Logger.info('Create stream.');
     await this._quicTransport.ready;
     // TODO: Potential failure because of publication stream is created faster
@@ -129,6 +137,28 @@ export class QuicConnection extends EventDispatcher {
     return quicStream;
   }
 
+  async publish(stream) {
+    // TODO: Avoid a stream to be published twice. The first 16 bit data send to
+    // server must be it's publication ID.
+    // TODO: Potential failure because of publication stream is created faster
+    // than signaling stream(created by the 1st call to initiatePublication).
+    const publicationId = await this._initiatePublication();
+    const quicStream = stream.stream;
+    const writer = quicStream.writable.getWriter();
+    await writer.ready;
+    writer.write(this._uuidToUint8Array(publicationId));
+    writer.releaseLock();
+    Logger.info('publish id');
+    this._quicStreams.set(publicationId, quicStream);
+    const publication = new Publication(publicationId, () => {
+      this._signaling.sendSignalingMessage('unpublish', {id: publication})
+          .catch((e) => {
+            Logger.warning('MCU returns negative ack for unpublishing, ' + e);
+          });
+    } /* TODO: getStats, mute, unmute is not implemented */);
+    return publication;
+  }
+
   hasContentSessionId(id) {
     return this._quicStreams.has(id);
   }
@@ -154,29 +184,6 @@ export class QuicConnection extends EventDispatcher {
     return s;
   }
 
-  /**
-   * @function createStream
-   * @desc Create a bidirectional stream.
-   * @param {string} sessionId Publication or subscription ID.
-   * @private
-   */
-  createStream(sessionId) {
-    Logger.info('Create stream.');
-    if (this._quicTransport && this._quicTransport.state == 'connected') {
-      this._quicStreams.push(stream);
-      return Promise.resolve(stream);
-    }
-    if (!this._quicTransport) {
-      this._initialize();
-    }
-    return new Promise((resolve, reject) => {
-      this._createStreamPromise = {
-        resolve: resolve,
-        reject: reject
-      };
-    });
-  }
-
   subscribe(stream) {
     const p = new Promise((resolve, reject) => {
       this._signaling
@@ -212,13 +219,13 @@ export class QuicConnection extends EventDispatcher {
     });
   }
 
-  _updateTransportId(id) {
+  async _updateTransportId(id) {
     if (this._transportId) {
       Logger.error('Update transport ID is not supported.');
       return;
     }
     this._transportId = id;
-    this._authenticate(id);
+    await this._authenticate(id);
   }
 
   async _initiatePublication() {
@@ -228,7 +235,7 @@ export class QuicConnection extends EventDispatcher {
       transport: {type: 'quic', id: this._transportId},
     });
     if (data.transportId) {
-      this._updateTransportId(data.transportId);
+      await this._updateTransportId(data.transportId);
     } else {
       if (this._transportId !== data.transportId) {
         throw new Error('Transport ID not match.');