Fix streaming-in internal spreading (#1082)

starwarfan · web-flow · commit 357b000194ca · 2021-09-09T14:34:40.000+08:00
Fix streaming-in internal spreading
diff --git a/source/agent/analytics/log4cxx.properties b/source/agent/analytics/log4cxx.properties
@@ -11,8 +11,13 @@ log4j.appender.A1.layout.ConversionPattern=%d  - %p: %c - %m%n
 
 # The raw UDP and TCP transports which are used for the connection between the Gateway and AVS.
 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.owt.SctpTransport=INFO
+log4j.logger.owt.InternalClient=INFO
+log4j.logger.owt.InternalServer=INFO
 
 log4j.logger.mcu.media.VideoMixer=INFO
 log4j.logger.mcu.media.VideoTranscoder=INFO
diff --git a/source/agent/analytics/log4js_configuration.json b/source/agent/analytics/log4js_configuration.json
@@ -18,6 +18,6 @@
     "AnalyticsNode": "INFO",
     "AnalyticsAgent": "WARN",
     "Connections": "INFO",
-    "InternalConnectionFactory": "INFO"
+    "InternalConnectionRouter": "INFO"
   }
 }
diff --git a/source/agent/audio/log4cxx.properties b/source/agent/audio/log4cxx.properties
@@ -16,6 +16,9 @@ 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.owt.SctpTransport=INFO
+log4j.logger.owt.InternalClient=INFO
+log4j.logger.owt.InternalServer=INFO
+
 log4j.logger.owt.AudioRanker=INFO
 
 log4j.logger.mcu.media.AudioMixer=INFO
diff --git a/source/agent/audio/log4js_configuration.json b/source/agent/audio/log4js_configuration.json
@@ -19,6 +19,6 @@
     "SelectiveMixer": "INFO",
     "WorkingAgent": "INFO",
     "Connections": "INFO",
-    "InternalConnectionFactory": "INFO"
+    "InternalConnectionRouter": "INFO"
   }
 }
diff --git a/source/agent/quic/log4js_configuration.json b/source/agent/quic/log4js_configuration.json
@@ -19,7 +19,7 @@
     "Connection": "WARN",
     "WorkingAgent": "WARN",
     "Connections": "WARN",
-    "InternalConnectionFactory": "WARN",
+    "InternalConnectionRouter": "WARN",
     "QuicConnection": "WARN",
     "QuicTransportServer": "WARN",
     "QuicTransportStreamPipeline": "WARN"
diff --git a/source/agent/recording/dist.json b/source/agent/recording/dist.json
@@ -47,6 +47,10 @@
                 "folder": "../addons/logger",
                 "name": "logger"
             },
+            {
+                "folder": "../addons/mediaFrameMulticaster",
+                "name": "mediaFrameMulticaster"
+            },
             {
                 "folder": "../addons/quicIO",
                 "name": "quicIO"
diff --git a/source/agent/recording/index.js b/source/agent/recording/index.js
@@ -11,6 +11,10 @@ var InternalOut = internalIO.Out;
 var avstream = require('../avstreamLib/build/Release/avstream');
 var AVStreamIn = avstream.AVStreamIn;
 var AVStreamOut = avstream.AVStreamOut;
+
+const MediaFrameMulticaster = require(
+    '../mediaFrameMulticaster/build/Release/mediaFrameMulticaster');
+
 var logger = require('../logger').logger;
 var path = require('path');
 var Connections = require('./connections');
@@ -42,6 +46,19 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         setTimeout(function () {connection.startPlay();}, 6000);
         callback('onStatus', {type: 'ready'});
 
+        var dispatcher = new MediaFrameMulticaster();
+        connection.addDestination('audio', dispatcher);
+        connection.addDestination('video', dispatcher);
+        connection.selfClose = connection.close;
+        connection.close = function () {
+            connection.removeDestination('audio', dispatcher);
+            connection.removeDestination('video', dispatcher);
+            connection.selfClose();
+        };
+        connection.source = function () {
+            return dispatcher.source();
+        };
+
         return connection;
     };
 
@@ -116,7 +133,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             return callback('callback', {type: 'failed', reason: 'Create Connection failed'});
         }
 
-        router.addLocalSource(connectionId, connectionType, conn)
+        router.addLocalSource(connectionId, connectionType, conn.source())
         .then(onSuccess(callback), onError(callback));
     };
 
diff --git a/source/agent/recording/log4cxx.properties b/source/agent/recording/log4cxx.properties
@@ -16,6 +16,8 @@ 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.owt.SctpTransport=INFO
+log4j.logger.owt.InternalClient=INFO
+log4j.logger.owt.InternalServer=INFO
 
 log4j.logger.owt.LiveStreamIn=INFO
 log4j.logger.owt.LiveStreamIn.JitterBuffer=INFO
diff --git a/source/agent/recording/log4js_configuration.json b/source/agent/recording/log4js_configuration.json
@@ -18,6 +18,6 @@
     "RecordingNode": "INFO",
     "WorkingAgent": "INFO",
     "Connections": "INFO",
-    "InternalConnectionFactory": "INFO"
+    "InternalConnectionRouter": "INFO"
   }
 }
diff --git a/source/agent/streaming/dist.json b/source/agent/streaming/dist.json
@@ -47,6 +47,10 @@
                 "folder": "../addons/logger",
                 "name": "logger"
             },
+            {
+                "folder": "../addons/mediaFrameMulticaster",
+                "name": "mediaFrameMulticaster"
+            },
             {
                 "folder": "../addons/quicIO",
                 "name": "quicIO"
diff --git a/source/agent/streaming/index.js b/source/agent/streaming/index.js
@@ -11,6 +11,10 @@ var InternalOut = internalIO.Out;
 var avstream = require('../avstreamLib/build/Release/avstream');
 var AVStreamIn = avstream.AVStreamIn;
 var AVStreamOut = avstream.AVStreamOut;
+
+const MediaFrameMulticaster = require(
+    '../mediaFrameMulticaster/build/Release/mediaFrameMulticaster');
+
 var logger = require('../logger').logger;
 var path = require('path');
 var Connections = require('./connections');
@@ -45,6 +49,19 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             notifyStatus(options.controller, sessionId, 'in', JSON.parse(message));
         });
 
+        var dispatcher = new MediaFrameMulticaster();
+        connection.addDestination('audio', dispatcher);
+        connection.addDestination('video', dispatcher);
+        connection.selfClose = connection.close;
+        connection.close = function () {
+            connection.removeDestination('audio', dispatcher);
+            connection.removeDestination('video', dispatcher);
+            connection.selfClose();
+        };
+        connection.source = function () {
+            return dispatcher.source();
+        };
+
         return connection;
     };
 
@@ -126,7 +143,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
             return callback('callback', {type: 'failed', reason: 'Create Connection failed'});
         }
 
-        router.addLocalSource(connectionId, connectionType, conn)
+        router.addLocalSource(connectionId, connectionType, conn.source())
         .then(onSuccess(callback), onError(callback));
     };
 
diff --git a/source/agent/streaming/log4cxx.properties b/source/agent/streaming/log4cxx.properties
@@ -16,6 +16,8 @@ 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.owt.SctpTransport=INFO
+log4j.logger.owt.InternalClient=INFO
+log4j.logger.owt.InternalServer=INFO
 
 log4j.logger.owt.LiveStreamIn=INFO
 log4j.logger.owt.LiveStreamIn.JitterBuffer=INFO
diff --git a/source/agent/streaming/log4js_configuration.json b/source/agent/streaming/log4js_configuration.json
@@ -18,6 +18,6 @@
     "StreamingNode": "INFO",
     "WorkingAgent": "INFO",
     "Connections": "INFO",
-    "InternalConnectionFactory": "INFO"
+    "InternalConnectionRouter": "INFO"
   }
 }
diff --git a/source/agent/video/log4cxx.properties b/source/agent/video/log4cxx.properties
@@ -16,6 +16,8 @@ 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.owt.SctpTransport=INFO
+log4j.logger.owt.InternalClient=INFO
+log4j.logger.owt.InternalServer=INFO
 
 log4j.logger.mcu.media.VideoMixer=INFO
 log4j.logger.mcu.media.VideoTranscoder=INFO
diff --git a/source/agent/video/log4js_configuration.json b/source/agent/video/log4js_configuration.json
@@ -19,6 +19,6 @@
     "VideoNode": "INFO",
     "WorkingAgent": "INFO",
     "Connections": "INFO",
-    "InternalConnectionFactory": "INFO"
+    "InternalConnectionRouter": "INFO"
   }
 }
diff --git a/source/agent/webrtc/log4js_configuration.json b/source/agent/webrtc/log4js_configuration.json
@@ -22,6 +22,6 @@
     "SdpInfo": "INFO",
     "WorkingAgent": "INFO",
     "Connections": "INFO",
-    "InternalConnectionFactory": "INFO"
+    "InternalConnectionRouter": "INFO"
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,6 @@`
`18`	`18`	`"AnalyticsNode": "INFO",`
`19`	`19`	`"AnalyticsAgent": "WARN",`
`20`	`20`	`"Connections": "INFO",`
`21`		`- "InternalConnectionFactory": "INFO"`
	`21`	`+ "InternalConnectionRouter": "INFO"`
`22`	`22`	`}`
`23`	`23`	`}`
Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,6 @@`
`19`	`19`	`"SelectiveMixer": "INFO",`
`20`	`20`	`"WorkingAgent": "INFO",`
`21`	`21`	`"Connections": "INFO",`
`22`		`- "InternalConnectionFactory": "INFO"`
	`22`	`+ "InternalConnectionRouter": "INFO"`
`23`	`23`	`}`
`24`	`24`	`}`