Add data source for linkup.

Author: jianjunz

source/agent/analytics/index.js

@@ -132,9 +132,9 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
         .then(rpcSuccess(callback))
         .catch(rpcError(callback));
     },
-    linkup: function (connectionId, audioFrom, videoFrom, callback) {
+    linkup: function (connectionId, audioFrom, videoFrom, dataFrom, callback) {
       log.debug('linkup,','connectionId:',connectionId,'videoFrom:',videoFrom,'callback:',callback);
-      agent.linkup(connectionId, audioFrom, videoFrom)
+      agent.linkup(connectionId, audioFrom, videoFrom, dataFrom)
         .then(rpcSuccess(callback))
         .catch(rpcError(callback));
     },

source/agent/audio/index.js

@@ -252,7 +252,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         }
     };
 
-    that.linkup = function (connectionId, audio_stream_id, video_stream_id, callback) {
+    that.linkup = function (connectionId, audio_stream_id, video_stream_id, data_stream_id, callback) {
         log.debug('linkup, connectionId:', connectionId, 'audio_stream_id:', audio_stream_id);
         if (connections[connectionId] === undefined) {
             return callback('callback', 'error', 'connection does not exist');

source/agent/base-agent.js

@@ -132,10 +132,11 @@ class BaseAgent {
     *@param {string} connectionId
     *@param {string} audioFrom
     *@param {string} videoFrom
+    *@param {string} dataFrom
     */
-  linkup(connectionId, audioFrom, videoFrom) {
+  linkup(connectionId, audioFrom, videoFrom, dataFrom) {
     log.debug('linkup, connectionId:', connectionId);
-    return this.connections.linkupConnection(connectionId, audioFrom, videoFrom);
+    return this.connections.linkupConnection(connectionId, audioFrom, videoFrom, dataFrom);
   }
 
   /**

source/agent/conference/roomController.js

@@ -400,7 +400,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
             audio = ((streams[stream_id].audio && target_node_type !== 'vmixer' && target_node_type !== 'vxcoder') ? true : false),
             video = ((streams[stream_id].video && target_node_type !== 'amixer' && target_node_type !== 'axcoder') ? true : false),
             spread_id = stream_id + '@' + target_node;
-        const data = streams[stream_id].data;
+        const data = !!streams[stream_id].data;
 
         if (!audio && !video && !data) {
             return on_error('Cannot spread stream without audio, video or data.');
@@ -491,6 +491,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
                             publisher: publisher,
                             audio: (audio ? {codec: streams[stream_id].audio.format} : false),
                             video: (video ? {codec: streams[stream_id].video.format} : false),
+                            data: data,
                             ip: from.ip,
                             port: from.port,
                         }
@@ -520,15 +521,15 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
             log.debug('internally publish/subscribe ok');
 
             // Linkup after publish/subscribe ready
-            return new Promise(function (resolve, reject) {
+            return new Promise(function(resolve, reject) {
                 makeRPC(
                     rpcClient,
                     original_node,
                     'linkup',
-                    [spread_id, audio ? stream_id : undefined, video ? stream_id : undefined],
+                    [ spread_id, audio ? stream_id : undefined, video ? stream_id : undefined, data ? stream_id : undefined ],
                     resolve,
                     reject);
-                });
+            });
         }).then(function () {
             if (streams[stream_id]) {
                 log.debug('internally linkup ok');

source/agent/quic/index.js

@@ -12,7 +12,7 @@
 // or |undefined|, which means QuicStream is not ready at this moment. It also
 // allows updating its associated QuicStream to a new one.
 // publish, unpublish, subscribe, unsubscribe, linkup, cutoff are required by
-// all agents. AFAIK, there is no documentation about these interfaces.
+// all agents. They are defined in base-agent.js.
 
 'use strict';
 const Connections = require('./connections');

source/agent/recording/index.js

@@ -184,7 +184,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         }, onError(callback));
     };
 
-    that.linkup = function (connectionId, audioFrom, videoFrom, callback) {
+    that.linkup = function (connectionId, audioFrom, videoFrom, dataFrom, callback) {
         connections.linkupConnection(connectionId, audioFrom, videoFrom).then(onSuccess(callback), onError(callback));
     };
 

source/agent/sip/index.js

@@ -818,7 +818,7 @@ module.exports = function (rpcC, selfRpcId, parentRpcId, clusterWorkerIP) {
         }
     };
 
-    that.linkup = function (connectionId, audioFrom, videoFrom, callback) {
+    that.linkup = function (connectionId, audioFrom, videoFrom, dataFrom, callback) {
         log.debug('linkup, connectionId:', connectionId, ', audioFrom:', audioFrom, ', videoFrom:', videoFrom);
 
         if (audioFrom && streams[audioFrom] === undefined) {

source/agent/streaming/index.js

@@ -195,7 +195,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         }, onError(callback));
     };
 
-    that.linkup = function (connectionId, audioFrom, videoFrom, callback) {
+    that.linkup = function (connectionId, audioFrom, videoFrom, dataFrom, callback) {
         connections.linkupConnection(connectionId, audioFrom, videoFrom).then(onSuccess(callback), onError(callback));
     };
 

source/agent/video/index.js

@@ -569,7 +569,7 @@ function VMixer(rpcClient, clusterIP) {
         }
     };
 
-    that.linkup = function (connectionId, audio_stream_id, video_stream_id, callback) {
+    that.linkup = function (connectionId, audio_stream_id, video_stream_id, data_stream_id, callback) {
         log.debug('linkup, connectionId:', connectionId, 'video_stream_id:', video_stream_id);
         if (connections[connectionId] === undefined) {
             return callback('callback', 'error', 'connection does not exist');
@@ -1028,7 +1028,7 @@ function VTranscoder(rpcClient, clusterIP) {
         }
     };
 
-    that.linkup = function (connectionId, audio_stream_id, video_stream_id, callback) {
+    that.linkup = function (connectionId, audio_stream_id, video_stream_id, data_stream_id, callback) {
         log.debug('linkup, connectionId:', connectionId, 'video_stream_id:', video_stream_id);
         if (connections[connectionId] === undefined) {
             return callback('callback', 'error', 'connection does not exist');

source/agent/webrtc/index.js

@@ -312,8 +312,8 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         }
     };
 
-    that.linkup = function (connectionId, audioFrom, videoFrom, callback) {
-        log.debug('linkup, connectionId:', connectionId, 'audioFrom:', audioFrom, 'videoFrom:', videoFrom);
+    that.linkup = function (connectionId, audioFrom, videoFrom, dataFrom, callback) {
+        log.debug('linkup, connectionId:', connectionId, 'audioFrom:', audioFrom, 'videoFrom:', videoFrom, 'callback:', callback);
         connections.linkupConnection(connectionId, audioFrom, videoFrom).then(onSuccess(callback), onError(callback));
     };