Add support to subscribe data stream.

Author: jianjunz

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

@@ -8,6 +8,21 @@ let quicChannel = null;
 let bidirectionalStream = null;
 let writeTask;
 const conference=new Owt.Conference.ConferenceClient();
+conference.addEventListener('streamadded', async (event) => {
+  console.log(event.stream);
+  if (event.stream.source.data) {
+    const subscription = await conference.subscribe(event.stream);
+    const reader = subscription.stream.readable.getReader();
+    while (true) {
+      const {value, done} = await reader.read();
+      if (done) {
+        console.log('Subscription ends.');
+        break;
+      }
+      console.log('Received data: '+value);
+    }
+  }
+});
 
 function updateConferenceStatus(message) {
   document.getElementById('conference-status').innerHTML +=

src/sdk/base/stream.js

@@ -25,10 +25,12 @@ function isAllowedValue(obj, allowedValues) {
  * "mic", "screen-cast", "file", "mixed" or undefined.
  * @param {?string} videoSourceInfo Video source info. Accepted values are:
  * "camera", "screen-cast", "file", "mixed" or undefined.
+ * @param {boolean} dataSourceInfo Indicates whether it is data. Accepted values
+ * are boolean.
  */
 export class StreamSourceInfo {
   // eslint-disable-next-line require-jsdoc
-  constructor(audioSourceInfo, videoSourceInfo) {
+  constructor(audioSourceInfo, videoSourceInfo, dataSourceInfo) {
     if (!isAllowedValue(audioSourceInfo, [undefined, 'mic', 'screen-cast',
       'file', 'mixed'])) {
       throw new TypeError('Incorrect value for audioSourceInfo');
@@ -39,6 +41,7 @@ export class StreamSourceInfo {
     }
     this.audio = audioSourceInfo;
     this.video = videoSourceInfo;
+    this.data = dataSourceInfo;
   }
 }
 /**

src/sdk/conference/client.js

@@ -281,9 +281,11 @@ export const ConferenceClient = function(config, signalingImpl) {
       if (streamInfo.media.video) {
         videoSourceInfo = streamInfo.media.video.source;
       }
+      const dataSourceInfo = streamInfo.data;
       const stream = new StreamModule.RemoteStream(streamInfo.id,
-          streamInfo.info.owner, undefined, new StreamModule.StreamSourceInfo(
-              audioSourceInfo, videoSourceInfo), streamInfo.info.attributes);
+        streamInfo.info.owner, undefined, new StreamModule.StreamSourceInfo(
+          audioSourceInfo, videoSourceInfo, dataSourceInfo), streamInfo.info
+        .attributes);
       stream.settings = StreamUtilsModule.convertToPublicationSettings(
           streamInfo.media);
       stream.extraCapabilities = StreamUtilsModule
@@ -434,12 +436,14 @@ export const ConferenceClient = function(config, signalingImpl) {
     if (!(stream instanceof StreamModule.RemoteStream)) {
       return Promise.reject(new ConferenceError('Invalid stream.'));
     }
-    if(!stream.source.audio&&!stream.source.video){ // QUIC stream.
-      const dataChannel=createDataChannel();
-      dataChannel.addEventListener('id',messageEvent=>{
-        incomingDataChannels.set(messageEvent.message, dataChannel);
-      });
-      return dataChannel.subscribe(stream);
+    if (stream.source.data) {
+      if (stream.source.audio || stream.source.video) {
+        return Promise.reject(new TypeError(
+          'Invalid source info. A remote stream is either a data stream or a media stream.'
+          ));
+      }
+      return quicTransportChannel.subscribe(stream);
+      // TODO
     }
     const channel = createPeerConnectionChannel(options.transport);
     return channel.subscribe(stream, options);
@@ -484,7 +488,7 @@ export const ConferenceClient = function(config, signalingImpl) {
    */
   this.createQuicConnection = function() {
     const quicConnection = new QuicConnection(
-      'quic-transport://jianjunz-nuc-ubuntu.sh.intel.com:7700/echo',
+      'quic-transport://example.com', me.id,
       createSignalingForChannel());
     quicTransportChannel=quicConnection;
     return quicConnection;

src/sdk/conference/quicconnection.js

@@ -12,6 +12,7 @@ import {
   EventDispatcher,
   MessageEvent,
 } from '../base/event.js';
+import {Subscription} from './subscription.js';
 
 /**
  * @class QuicConnection
@@ -20,12 +21,14 @@ import {
  * @private
  */
 export class QuicConnection extends EventDispatcher {
-  constructor(url, signaling) {
+  constructor(url, token, signaling) {
     super();
     this._signaling = signaling;
     this._ended = false;
     this._quicStreams = new Map(); // Key is publication or subscription ID.
     this._quicTransport = new QuicTransport(url);
+    this._subscribePromises = new Map(); // Key is subscription ID.
+    this._init(token);
   }
 
   /**
@@ -54,23 +57,81 @@ export class QuicConnection extends EventDispatcher {
     }
   }
 
+  async _init(token) {
+    this._authenticate(token);
+    const receiveStreamReader =
+        this._quicTransport.receiveStreams().getReader();
+    Logger.info('Reader: '+receiveStreamReader);
+    while (true) {
+      const {
+        value: receiveStream,
+        done: readingReceiveStreamsDone
+      } = await receiveStreamReader.read();
+      Logger.info('New stream received');
+      if (readingReceiveStreamsDone) {
+        break;
+      }
+      const chunkReader = receiveStream.readable.getReader();
+      const {
+        value: uuid,
+        done: readingChunksDone
+      } = await chunkReader.read();
+      if (readingChunksDone) {
+        Logger.error('Stream closed unexpectedly.');
+        return;
+      }
+      if (uuid.length != 16) {
+        Logger.error('Unexpected length for UUID.');
+        return;
+      }
+      chunkReader.releaseLock();
+      const subscriptionId = this._uint8ArrayToUuid(uuid);
+      this._quicStreams.set(subscriptionId, receiveStream);
+      if (this._subscribePromises.has(subscriptionId)) {
+        const subscription =
+            this._createSubscription(subscriptionId, receiveStream);
+        this._subscribePromises.get(subscriptionId).resolve(subscription);
+      }
+    }
+  }
+
+  _createSubscription(id, receiveStream) {
+    // TODO: Incomplete subscription.
+    const subscription = new Subscription(id, () => {
+      receiveStream.abortReading();
+    });
+    subscription.stream = receiveStream;
+    return subscription;
+  }
+
+  async _authenticate(token) {
+    await this._quicTransport.ready;
+    const quicStream = await this._quicTransport.createSendStream();
+    const writer = quicStream.writable.getWriter();
+    await writer.ready;
+    writer.write(new Uint8Array(16));
+    const encoder = new TextEncoder();
+    writer.write(encoder.encode(token));
+  }
+
   async createSendStream(sessionId) {
     Logger.info('Create stream.');
     await this._quicTransport.ready;
     // TODO: Creating quicStream and initializing publication concurrently.
     const quicStream = await this._quicTransport.createSendStream();
-    const publicationId = await this._initializePublication();
+    const publicationId = await this._initiatePublication();
     const writer= quicStream.writable.getWriter();
     await writer.ready;
-    writer.write(this.uuidToUint8Array(publicationId));
+    writer.write(this._uuidToUint8Array(publicationId));
+    writer.write(this._uuidToUint8Array(publicationId));
     this._quicStreams.set(publicationId, quicStream);
   }
 
   hasContentSessionId(id) {
     return this._quicStreams.has(id);
   }
 
-  uuidToUint8Array(uuidString) {
+  _uuidToUint8Array(uuidString) {
     if (uuidString.length != 32) {
       throw new TypeError('Incorrect UUID.');
     }
@@ -82,6 +143,15 @@ export class QuicConnection extends EventDispatcher {
     return uuidArray;
   }
 
+  _uint8ArrayToUuid(uuidBytes) {
+    let s = '';
+    for (let hex of uuidBytes) {
+      const str = hex.toString(16);
+      s += str.padStart(2, '0');
+    }
+    return s;
+  }
+
   /**
    * @function createStream
    * @desc Create a bidirectional stream.
@@ -106,23 +176,38 @@ export class QuicConnection extends EventDispatcher {
   }
 
   subscribe(stream) {
-    this._outgoing=false;
-    this._streamSubscribed = stream;
-    if (!this._quicTransport) {
-      this.createStream();
-    }
+    const p = new Promise((resolve, reject) => {
+      this._signaling
+          .sendSignalingMessage(
+              'subscribe',
+              {media: null, data: {from: stream.id}, transport: {type: 'quic'}})
+          .then((data) => {
+            if (this._quicStreams.has(data.id)) {
+              // QUIC stream created before signaling returns.
+              const subscription = this._createSubscription(
+                  data.id, this._quicStreams.get(data.id));
+              resolve(subscription);
+            } else {
+              // QUIC stream is not created yet, resolve promise after getting
+              // QUIC stream.
+              this._subscribePromises.set(
+                  data.id, {resolve: resolve, reject: reject});
+            }
+          });
+    });
+    return p;
   }
 
-  _readAndPrint(){
-    this._quicStreams[0].waitForReadable(5).then(()=>{
+  _readAndPrint() {
+    this._quicStreams[0].waitForReadable(5).then(() => {
       let data = new Uint8Array(this._quicStreams[0].readBufferedAmount);
       this._quicStreams[0].readInto(data);
       Logger.info('Read data: ' + data);
       this._readAndPrint();
     });
   }
 
-  async _initializePublication() {
+  async _initiatePublication() {
     const data = await this._signaling.sendSignalingMessage('publish', {
       media: null,
       data: true,