Fix QUIC agent for internal transport router.

A recent commit (773cfe4) for internal
transport introduced breaking changes. This commit updates QUIC agent
to work with updated internal transport APIs.

Author: jianjunz

source/agent/addons/quic/QuicTransportStream.cc

@@ -113,12 +113,12 @@ NAN_METHOD(QuicTransportStream::close)
 NAN_METHOD(QuicTransportStream::addDestination)
 {
     QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
-    if (info.Length() != 1) {
+    if (info.Length() != 2) {
         Nan::ThrowTypeError("Invalid argument length for addDestination.");
         return;
     }
     // TODO: Check if info[0] is an Nan wrapped object.
-    auto framePtr = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info[0]->ToObject());
+    auto framePtr = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info[1]->ToObject());
     // void* ptr = info[0]->ToObject()->GetAlignedPointerFromInternalField(0);
     // auto framePtr=static_cast<owt_base::FrameDestination*>(ptr);
     obj->addDataDestination(framePtr);

source/agent/quic/dist.json

@@ -29,7 +29,8 @@
               "../connections.js",
               "../InternalConnectionFactory.js",
               "../../common/makeRPC.js",
-              "../../common/rpcChannel.js"
+              "../../common/rpcChannel.js",
+              "../internalConnectionRouter.js"
           ],
           "quic/webtransport": [
               "webtransport/quicTransportServer.js",

source/agent/quic/index.js

@@ -15,7 +15,6 @@
 // all agents. They are defined in base-agent.js.
 
 'use strict';
-const Connections = require('./connections');
 const InternalConnectionFactory = require('./InternalConnectionFactory');
 const logger = require('../logger').logger;
 const QuicTransportServer = require('./webtransport/quicTransportServer');
@@ -25,6 +24,7 @@ const log = logger.getLogger('QuicNode');
 const addon = require('./build/Release/quic');
 const cipher = require('../cipher');
 const path = require('path');
+const {InternalConnectionRouter} = require('./internalConnectionRouter');
 
 log.info('QUIC transport node.')
 
@@ -34,7 +34,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
       agentID: parentRpcId,
       clusterIP: clusterWorkerIP
     };
-    const connections = new Connections;
+    const router = new InternalConnectionRouter(global.config.internal);
     const internalConnFactory = new InternalConnectionFactory;
     const incomingStreamPipelines =
         new Map();  // Key is publication ID, value is stream pipeline.
@@ -100,18 +100,22 @@ 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.
+        log.debug(
+            'A stream with session ID ' + stream.contentSessionId +
+            ' is added.');
+        let pipeline = null;
+        if (outgoingStreamPipelines.has(stream.contentSessionId)) {
+          pipeline = outgoingStreamPipelines.get(stream.contentSessionId);
+        } else if (incomingStreamPipelines.has(stream.contentSessionId)) {
+          pipeline = incomingStreamPipelines.get(stream.contentSessionId);
         } else {
           log.warn(
               'Cannot find a pipeline for QUIC stream. Content session ID: ' +
               stream.contentSessionId);
           stream.close();
+          return;
         }
+        pipeline.quicStream(stream);
       });
       quicTransportServer.on('connectionadded', (connection) => {
         log.debug(
@@ -159,32 +163,21 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         };
     };
 
-    that.createInternalConnection = function (connectionId, direction, internalOpt, callback) {
-        internalOpt.minport = global.config.internal.minport;
-        internalOpt.maxport = global.config.internal.maxport;
-        var portInfo = internalConnFactory.create(connectionId, direction, internalOpt);
-        callback('callback', {ip: that.clusterIP, port: portInfo});
-    };
-
-    that.destroyInternalConnection = function (connectionId, direction, callback) {
-        internalConnFactory.destroy(connectionId, direction);
-        callback('callback', 'ok');
+    that.getInternalAddress = function(callback) {
+        const ip = global.config.internal.ip_address;
+        const port = router.internalPort;
+        callback('callback', {ip, port});
     };
 
     // functions: publish, unpublish, subscribe, unsubscribe, linkup, cutoff
     that.publish = function(connectionId, connectionType, options, callback) {
         log.debug('publish, connectionId:', connectionId, 'connectionType:', connectionType, 'options:', options);
-        if (connections.getConnection(connectionId)) {
+        if (router.getConnection(connectionId)) {
             return callback('callback', {type: 'failed', reason: 'Connection already exists:'+connectionId});
         }
 
         var conn = null;
         switch (connectionType) {
-        case 'internal':
-            conn = internalConnFactory.fetch(connectionId, 'in');
-            if (conn)
-                conn.connect(options);
-            break;
         case 'quic':
             conn = createStreamPipeline(connectionId, 'in', options, callback);
             if (!conn) {
@@ -198,14 +191,16 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             log.error('Create connection failed', connectionId, connectionType);
             return callback('callback', {type: 'failed', reason: 'Create Connection failed'});
         }
-        connections.addConnection(connectionId, connectionType, options.controller, conn, 'in')
-        .then(onSuccess(callback), onError(callback));
+        conn.bindRouterAsSourceCallback = function(stream) {
+          router.addLocalSource(connectionId, connectionType, stream);
+        }
+        onSuccess(callback)();
     };
 
     that.unpublish = function (connectionId, callback) {
         log.debug('unpublish, connectionId:', connectionId);
-        var conn = connections.getConnection(connectionId);
-        connections.removeConnection(connectionId).then(function(ok) {
+        var conn = router.getConnection(connectionId);
+        router.removeConnection(connectionId).then(function(ok) {
             if (conn && conn.type === 'internal') {
                 internalConnFactory.destroy(connectionId, 'in');
             } else if (conn) {
@@ -225,17 +220,12 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         if(!options.data){
             log.error('Subscription request does not include data field.');
         }
-        if (connections.getConnection(connectionId)) {
+        if (router.getConnection(connectionId)) {
             return callback('callback', {type: 'failed', reason: 'Connection already exists:'+connectionId});
         }
 
         var conn = null;
         switch (connectionType) {
-            case 'internal':
-                conn = internalConnFactory.fetch(connectionId, 'out');
-                if (conn)
-                    conn.connect(options);//FIXME: May FAIL here!!!!!
-                break;
             case 'quic':
                 conn = createStreamPipeline(connectionId, 'out', options, callback);
                 const stream = quicTransportServer.createSendStream(options.transport.id, connectionId);
@@ -252,14 +242,14 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             return callback('callback', {type: 'failed', reason: 'Create Connection failed'});
         }
 
-        connections.addConnection(connectionId, connectionType, options.controller, conn, 'out')
+        router.addLocalDestination(connectionId, connectionType, conn)
         .then(onSuccess(callback), onError(callback));
     };
 
     that.unsubscribe = function (connectionId, callback) {
         log.debug('unsubscribe, connectionId:', connectionId);
-        var conn = connections.getConnection(connectionId);
-        connections.removeConnection(connectionId).then(function(ok) {
+        var conn = router.getConnection(connectionId);
+        router.removeConnection(connectionId).then(function(ok) {
             if (conn && conn.type === 'internal') {
                 internalConnFactory.destroy(connectionId, 'out');
             } else if (conn) {
@@ -269,19 +259,19 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         }, onError(callback));
     };
 
-    that.linkup = function (connectionId, audioFrom, videoFrom, dataFrom, callback) {
-        log.debug('linkup.');
-        connections.linkupConnection(connectionId, audioFrom, videoFrom, dataFrom).then(onSuccess(callback), onError(callback));
+    that.linkup = function (connectionId, from, callback) {
+        log.debug('linkup, connectionId:', connectionId, 'from:', from);
+        router.linkup(connectionId, from).then(onSuccess(callback), onError(callback));
     };
 
     that.cutoff = function (connectionId, callback) {
         log.debug('cutoff, connectionId:', connectionId);
-        connections.cutoffConnection(connectionId).then(onSuccess(callback), onError(callback));
+        router.cutoff(connectionId).then(onSuccess(callback), onError(callback));
     };
 
     that.mediaOnOff = function (connectionId, track, direction, action, callback) {
         log.debug('mediaOnOff, connection id:', connectionId, 'track:', track, 'direction:', direction, 'action:', action);
-        var conn = connections.getConnection(connectionId);
+        var conn = router.getConnection(connectionId);
         if (conn) {
             if (conn.type === 'quic') {//NOTE: Only webrtc connection supports media-on-off
                 conn.connection.onTrackControl(track,
@@ -304,21 +294,11 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
     };
 
     that.close = function() {
-        log.debug('close called');
-        var connIds = connections.getIds();
-        for (let connectionId of connIds) {
-            var conn = connections.getConnection(connectionId);
-            connections.removeConnection(connectionId);
-            if (conn && conn.type === 'internal') {
-                internalConnFactory.destroy(connectionId, conn.direction);
-            } else if (conn && conn.connection) {
-                conn.connection.close();
-            }
-        }
+        router.clear();
     };
 
     that.onFaultDetected = function (message) {
-        connections.onFaultDetected(message);
+        router.onFaultDetected(message);
     };
 
     return that;

source/agent/quic/log4cxx.properties

@@ -12,17 +12,20 @@ log4j.appender.A1.layout.ConversionPattern=%d  - %p: %c - %m%n
 # The utility to handle the received RTP packets from the browser,
 # to handle the FEC/RED packets, generate NACK requests upon packet loss, and
 # count receive statistics (packet loss etc.) for RTCP Receiver feedback generation, etc.
-log4j.logger.woogeen.ProtectedRTPReceiver=INFO
+log4j.logger.owt.ProtectedRTPReceiver=INFO
 
 # The utility to handle the RTP packets sent to the browser,
 # to construct the FEC/RED packets if necessary, resend lost packets, and
 # count send statistics (bitrate etc.) for internal QoS parameters adjustment, etc.
-log4j.logger.woogeen.ProtectedRTPSender=INFO
+log4j.logger.owt.ProtectedRTPSender=INFO
 
 # The raw UDP and TCP transports which are used for the connection between the Gateway and AVS.
-log4j.logger.woogeen.RawTransport=INFO
+log4j.logger.owt.RawTransport=INFO
+log4j.logger.owt.TransportSession=INFO
+log4j.logger.owt.TransportServer=INFO
+log4j.logger.owt.TransportClient=INFO
 # If the SctpTransport log is set to debug, heavy IO would affact the connections
-log4j.logger.woogeen.SctpTransport=INFO
+log4j.logger.owt.SctpTransport=INFO
 
 # The VideoFeedbackReactor is used to connect the WebRTCFeedbackProcessor and
 # the ProtectedRTPSender; the WebRTCFeedbackProcessor handles the received RTCP
@@ -31,20 +34,20 @@ log4j.logger.woogeen.SctpTransport=INFO
 # information to the VideoFeedbackReactor, which will adjust the (FEC) protection parameters
 # according to the RTT, the target send bitrate and the actual bitrate from the
 # ProtectedRTPSender statistics, and notify ProtectedRTPSender about the new parameters.
-log4j.logger.woogeen.VideoFeedbackReactor=INFO
-log4j.logger.woogeen.WebRTCFeedbackProcessor=INFO
+log4j.logger.owt.VideoFeedbackReactor=INFO
+log4j.logger.owt.WebRTCFeedbackProcessor=INFO
 
-log4j.logger.woogeen.AudioFrameConstructor=INFO
-log4j.logger.woogeen.AudioFramePacketizer=INFO
+log4j.logger.owt.AudioFrameConstructor=INFO
+log4j.logger.owt.AudioFramePacketizer=INFO
 
-log4j.logger.woogeen.VideoFrameConstructor=INFO
-log4j.logger.woogeen.VideoFramePacketizer=INFO
+log4j.logger.owt.VideoFrameConstructor=INFO
+log4j.logger.owt.VideoFramePacketizer=INFO
 
-log4j.logger.woogeen.LiveStreamIn=INFO
-log4j.logger.woogeen.LiveStreamIn.JitterBuffer=INFO
-log4j.logger.woogeen.AVStreamOut=INFO
-log4j.logger.woogeen.MediaFileOut=INFO
-log4j.logger.woogeen.LiveStreamOut=INFO
+log4j.logger.owt.LiveStreamIn=INFO
+log4j.logger.owt.LiveStreamIn.JitterBuffer=INFO
+log4j.logger.owt.AVStreamOut=INFO
+log4j.logger.owt.MediaFileOut=INFO
+log4j.logger.owt.LiveStreamOut=INFO
 
 log4j.logger.mcu.media.AudioMixer=INFO
 log4j.logger.mcu.media.AcmmFrameMixer=INFO
@@ -66,19 +69,19 @@ log4j.logger.mcu.media.SoftVideoCompositor.AvatarManager=INFO
 log4j.logger.mcu.media.SoftVideoCompositor.SoftInput=INFO
 log4j.logger.mcu.media.SoftVideoCompositor.SoftFrameGenerator=INFO
 
-log4j.logger.woogeen.VCMFrameDecoder=INFO
-log4j.logger.woogeen.VCMFrameEncoder=INFO
-log4j.logger.woogeen.SVTHEVCEncoder=INFO
-log4j.logger.woogeen.FrameProcesser=INFO
+log4j.logger.owt.VCMFrameDecoder=INFO
+log4j.logger.owt.VCMFrameEncoder=INFO
+log4j.logger.owt.SVTHEVCEncoder=INFO
+log4j.logger.owt.FrameProcesser=INFO
 
 # Msdk media pipeline
-log4j.logger.woogeen.MsdkBase=INFO
-log4j.logger.woogeen.MsdkFrame=INFO
-log4j.logger.woogeen.MsdkFramePool=INFO
-log4j.logger.woogeen.MsdkScaler=INFO
-log4j.logger.woogeen.MsdkFrameDecoder=INFO
-log4j.logger.woogeen.StreamEncoder=INFO
-log4j.logger.woogeen.MsdkFrameEncoder=INFO
+log4j.logger.owt.MsdkBase=INFO
+log4j.logger.owt.MsdkFrame=INFO
+log4j.logger.owt.MsdkFramePool=INFO
+log4j.logger.owt.MsdkScaler=INFO
+log4j.logger.owt.MsdkFrameDecoder=INFO
+log4j.logger.owt.StreamEncoder=INFO
+log4j.logger.owt.MsdkFrameEncoder=INFO
 
 log4j.logger.mcu.media.MsdkVideoCompositor=INFO
 log4j.logger.mcu.media.MsdkVideoCompositor.MsdkAvatarManager=INFO

source/agent/quic/webtransport/quicTransportStreamPipeline.js

@@ -25,8 +25,11 @@ module.exports = class QuicTransportStreamPipeline {
     this._quicStream = null;
     this._notifiedReady = false;
 
-    this.quicStream = function (stream) {
+    this.quicStream = function(stream) {
       this._quicStream = stream;
+      if (this.bindRouterAsSourceCallback) {
+        this.bindRouterAsSourceCallback(stream);
+      }
       if (!this._notifiedReady) {
         updateStatus({
           type: 'ready',
@@ -39,7 +42,7 @@ module.exports = class QuicTransportStreamPipeline {
     };
 
     this.addDestination = function(name, dst) {
-      this._quicStream.addDestination(dst);
+      this._quicStream.addDestination(name, dst);
     };
 
     this.receiver = function(kind) {
@@ -50,8 +53,13 @@ module.exports = class QuicTransportStreamPipeline {
       return this._quicStream;
     };
 
-    this.close = function(){
+    this.close = function() {
       this._quicStream.close();
-    }
+    };
+
+    // https://github.com/open-webrtc-toolkit/owt-server/pull/988 changed the
+    // dataflow of frames. Since this object is not a native backed JavaScript
+    // object, router.addLocalSource should be called for this._quicStream.
+    this.bindRouterAsSourceCallback = null;
   }
 };