Fix publish method after rebasing.

Server can detect new streams from QUIC agent, but this change breaks
rtcController and stream events sent from server don't have "data"
property. These issues along with other subscribe issues will be fixed
in following commits.

Author: jianjunz

source/agent/conference/conference.js

@@ -397,7 +397,8 @@ var Conference = function (rpcClient, selfRpcId) {
                     const sessionInfo = {
                       locality: transport.locality,
                       media: media,
-                      info: { type: 'webrtc', owner: transport.owner }
+                      data: rtcInfo.data,
+                      info: { type: 'quic', owner: transport.owner }
                     };
                     sendMsgTo(transport.owner, 'progress', {id: transport.id, sessionId, status: 'ready'});
                     onSessionEstablished(transport.owner, sessionId, direction, sessionInfo);
@@ -949,7 +950,8 @@ var Conference = function (rpcClient, selfRpcId) {
       type: pubInfo.type,
       transportId: pubInfo.transportId,
       tracks: pubInfo.media.tracks,
-      legacy: pubInfo.legacy
+      legacy: pubInfo.legacy,
+      data: pubInfo.data
     };
     return rtcPubInfo;
   };
@@ -1013,7 +1015,7 @@ var Conference = function (rpcClient, selfRpcId) {
     }
 
     if (pubInfo.type === 'sip') {
-      return addStream(streamId, pubInfo.locality, pubInfo.transport, pubInfo.media, pubInfo.data, {owner: participantId, type: 'sip'})
+      return addStream(streamId, pubInfo.locality, pubInfo.transport, pubInfo.media, pubInfo.data, {owner: participantId, type: pubInfo.type})
       .then((result) => {
         callback('callback', result);
       })
@@ -1033,12 +1035,14 @@ var Conference = function (rpcClient, selfRpcId) {
     } else {
       var origin = participants[participantId].getOrigin();
       var format_preference;
-      if (pubInfo.type === 'webrtc') {
+      if (pubInfo.type === 'webrtc' || pubInfo.type === 'quic') {
         const rtcPubInfo = translateRtcPubIfNeeded(pubInfo);
-        // Set formatPreference
-        rtcPubInfo.tracks.forEach(track => {
-          track.formatPreference = {optional: room_config.mediaIn[track.type]};
-        });
+        if (rtcPubInfo.tracks) {
+            // Set formatPreference
+            rtcPubInfo.tracks.forEach(track => {
+                track.formatPreference = { optional : room_config.mediaIn[track.type] };
+            });
+        }
         initiateStream(streamId, {owner: participantId, type: pubInfo.type, origin});
         return rtcController.initiate(participantId, streamId, 'in', participants[participantId].getOrigin(), rtcPubInfo)
         .then((result) => {
@@ -1755,9 +1759,13 @@ var Conference = function (rpcClient, selfRpcId) {
     callback('callback', 'ok');
   };
 
-  that.onSessionProgress = function (sessionId, direction, sessionStatus) {
-    log.debug('onSessionProgress, sessionId:', sessionId, 'direction:', direction, 'sessionStatus:', sessionStatus);
-    accessController && accessController.onSessionStatus(sessionId, sessionStatus);
+  that.onSessionProgress = function(sessionId, direction, sessionStatus) {
+      log.debug('onSessionProgress, sessionId:', sessionId, 'direction:', direction, 'sessionStatus:', sessionStatus);
+      if (sessionStatus.data) {
+          rtcController.onSessionProgress(sessionId, sessionStatus);
+      } else {
+          accessController && accessController.onSessionStatus(sessionId, sessionStatus);
+      }
   };
 
   that.onTransportProgress = function (transportId, status) {

source/agent/conference/stream.js

@@ -256,7 +256,7 @@ class ForwardStream extends Stream {
         log.error(`Unexpected track type: ${track.type}`);
       }
     }
-    if (this.media.tracks.length === 0) {
+    if (!this.data && this.media.tracks.length === 0) {
       err = 'No valid tracks in stream';
     }
     return err;

source/agent/quic/index.js

@@ -45,12 +45,15 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         global.config.cluster.name :
         undefined;
 
+    const getConferenceController = async (roomId) => {
+      return await rpcChannel.makeRPC(clusterName, 'schedule', [
+        'conference', roomId, 'preference' /*TODO: specify preference*/,
+        30 * 1000
+      ]);
+    };
+
     const getRpcPortal = async (roomId, participantId) => {
-      const controllerAgent =
-          await rpcChannel.makeRPC(clusterName, 'schedule', [
-            'conference', roomId, 'preference' /*TODO: specify preference*/,
-            30 * 1000
-          ]);
+      const controllerAgent = await getConferenceController(roomId);
       let controller = null;
       let portal = null;
       const retry = 5;
@@ -97,17 +100,24 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
           password, validateToken);
       quicTransportServer.start();
       quicTransportServer.on('streamadded', (stream) => {
+        log.debug('A stream with session ID '+stream.contentSessionId+' is added.');
         const conn = connections.getConnection(stream.contentSessionId);
         if (conn) {
           // TODO: verify transport ID.
           conn.connection.quicStream(stream);
+          // TODO: Make RPC call to conference node for session-established.
         } else {
           log.warn(
               'Cannot find a pipeline for QUIC stream. Content session ID: ' +
               stream.contentSessionId);
           stream.close();
         }
       });
+      quicTransportServer.on('connectionadded', (connection) => {
+        log.debug(
+            'A connection for transport ID ' + connection.transportId +
+            ' is created.');
+      });
     });
 
     const createStreamPipeline =
@@ -176,7 +186,6 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
                 conn.connect(options);
             break;
         case 'quic':
-            quicTransportServer.addTransportId(options.transport.id);
             conn = createStreamPipeline(connectionId, 'in', options, callback);
             if (!conn) {
                 return;

source/agent/quic/webtransport/quicTransportServer.js

@@ -30,7 +30,6 @@ module.exports = class QuicTransportServer extends EventEmitter {
     this._streams = new Map(); // Key is content session ID.
     this._unAuthenticatedConnections = []; // When it's authenticated, it will be moved to this.connections.
     this._unAssociatedStreams = []; // No content session ID assgined to them.
-    this._assignedTransportIds = []; // Transport channels assigned to this server.
     this._validateTokenCallback = validateTokenCallback;
     this._server.onconnection = (connection) => {
       this._unAuthenticatedConnections.push(connection);
@@ -58,6 +57,9 @@ module.exports = class QuicTransportServer extends EventEmitter {
                 'A new stream ' + streamId + ' is created on transport ' +
                 connection.transportId);
             this.emit('streamadded', stream);
+            const uuidBytes =
+                this._uuidStringToUint8Array(connection.transportId);
+            stream.write(uuidBytes, uuidBytes.length);
           }
         };
         stream.ondata = (data) => {
@@ -113,9 +115,17 @@ module.exports = class QuicTransportServer extends EventEmitter {
                     connection.close();
                     return;
                   }
+                  log.debug('Created connection for transport ID '+connection.transportId);
                   connection.transportId = token.transportId;
+                  connection.participantId = token.participantId;
+                  // A user can join multiple rooms. And it would be better if a
+                  // connection can be shared by all these rooms. But it looks
+                  // like existing architecture doesn't support to do so. So we
+                  // bind a connection to a single room here.
+                  connection.roomId = token.roomId;
                   stream.transportId = token.transportId;
                   this._connections.set(connection.transportId, connection);
+                  this.emit('connectionadded', connection);
                 });
                 return;
               } else {  // Store the data.
@@ -135,12 +145,6 @@ module.exports = class QuicTransportServer extends EventEmitter {
     // TODO: Return error if server is not started, e.g.: port is not available.
   }
 
-  addTransportId(id) {
-    if (!(id in this._assignedTransportIds)) {
-      this._assignedTransportIds.push(id);
-    }
-  }
-
   // Create a send stream for specfied QuicTransport. If specified QuicTransport
   // doesn't exist, no stream will be created.
   createSendStream(transportId, contentSessionId) {

source/agent/quic/webtransport/quicTransportStreamPipeline.js

@@ -32,6 +32,7 @@ module.exports = class QuicTransportStreamPipeline {
           type: 'ready',
           audio: false,
           video: false,
+          data: true,
           simulcast: false
         });
       }
@@ -50,6 +51,7 @@ module.exports = class QuicTransportStreamPipeline {
     };
 
     this.close = function(){
+      return;
       this._quicStream.close();
     }
   }