Merge pull request #1134 from starwarfan/mst-st-layer

Add OWT stream layer control

Author: starwarfan

doc/design/adding-an-agent.md

@@ -0,0 +1,17 @@
+| Adding an agent in OWT |
+
+
+# Introduction
+
+OWT agent is an independent module which provides RPC interfaces for external callers. It has following characteristics:
+1. Use same RPC mechanism as other OWT modules.
+2. Register on OWT cluster manager as cluster worker.
+3. Spawn child process to handle actual tasks.
+
+
+# How to add an agent
+
+1. Create a directory under source, take `source/agent/sample` as an example.
+2. Create `package.json` in the new agent directory. This file plays a same role as other `package.json` in Node.js. Besides, we need define a `start` command in `scripts` which will be used when we start the new agent using default daemon script(e.g. bin/start-all.sh). The `start` command should have format as `node . -U [name]`. The [name] is also the entry file name of the new agent.
+3. Create `dist.json` in the new agent directory. This file defines packing rules for the new agent. See `source/agent/sample/dist.json` for reference.
+4. Create js files with new agent implementation following RPC-agent interfaces. The entry file should have same name as defined in start script of `package.json`. See `source/agent/sample/sample.js` for reference.

scripts/release/daemon-mcu.sh

@@ -78,6 +78,9 @@ mkdir -p "$LogDir"
 stdout=${LogDir}/${command}.stdout
 pid=${LogDir}/${command}.pid
 
+CommandDir=${command//-/_}
+StartCmd=""
+
 # Set default scheduling priority
 if [ "$OWT_NICENESS" = "" ]; then
     export OWT_NICENESS=0
@@ -95,6 +98,8 @@ case $startStop in
     check_node_version || exit 1
     rotate_log $stdout
     echo "starting $command, stdout -> $stdout"
+
+
     case ${command} in
       management-api )
         cd ${OWT_HOME}/management_api
@@ -212,8 +217,16 @@ case $startStop in
         echo $! > ${pid}
         ;;
       * )
-        echo $usage
-        exit 1
+        if [ -d ${OWT_HOME}/${CommandDir} ]; then
+          cd ${OWT_HOME}/${CommandDir}
+          StartCmd=$(node -e "process.stdout.write(require('./package.json').scripts.start)")
+          nohup nice -n ${OWT_NICENESS} ${StartCmd} \
+            > "${stdout}" 2>&1 </dev/null &
+          echo $! > ${pid}
+        else
+          echo $usage
+          exit 1
+        fi
         ;;
     esac
 

source/agent/agent.toml

@@ -0,0 +1,24 @@
+[agent]
+#Max processes that agent can run
+maxProcesses = -1 #default: -1, unlimited
+
+#Number of precesses that agent runs when it starts. 1 <= prerunProcesses <= maxProcesses.
+prerunProcesses = 2 #default: 2
+
+#Roles defined in services have been moved to database
+
+[cluster]
+name = "owt-cluster"
+
+#The number of times to retry joining if the first try fails.
+join_retry = 60 #default: 60
+
+#The interval of reporting the work load
+report_load_interval = 1000 #default: 1000, unit: millisecond
+
+#The max CPU load under which this worker can take new tasks.
+max_load = 0.85 #default: 0.85
+
+[rabbit]
+host = "localhost" #default: "localhost"
+port = 5672 #default: 5672

source/agent/analytics/index.js

@@ -66,6 +66,7 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
   }
 
   const notifyStatus = (controller, sessionId, direction, status) => {
+    log.warn('notifyStatus:', controller, sessionId, direction, status);
     rpcClient.remoteCast(controller, 'onSessionProgress', [sessionId, direction, status]);
     // Emit GRPC notifications
     const notification = {
@@ -257,7 +258,12 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
               return;
             }
           }
-          generateStream(options.controller, newStreamId, streamInfo);
+
+          // For Stream Engine, onSessionProgress(id, name, data)
+          streamInfo.id = newStreamId;
+          notifyStatus(controller, connectionId, 'onNewStream', streamInfo);
+
+          // generateStream(options.controller, newStreamId, streamInfo);
         } catch (e) {
           log.error("Parse stream added data with error:", e);
         }
@@ -313,7 +319,9 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
 
   that.cutoff = function (connectionId, callback) {
     log.debug('cutoff, connectionId:', connectionId);
-    router.cutoff(connectionId).then(onSuccess(callback), onError(callback));
+    router.cutoff(connectionId).then(
+      () => callback('callback', 'ok'),
+      () => callback('callback', 'error', 'Failed to cutoff'));
   };
 
   that.onFaultDetected = function (message) {

source/agent/audio/index.js

@@ -61,6 +61,10 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
 
     // For GRPC notifications
     var streamingEmitter = new EventEmitter();
+    // ConnectionId => InputId
+    const fromMap = new Map();
+    // InputId => options
+    const unlinkedInputs = new Map();
 
     var addInput = function (stream_id, owner, codec, options, on_ok, on_error) {
         if (engine) {
@@ -70,6 +74,13 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
                 port: options.port
             });
 
+            if (!conn) {
+                log.debug('addInput waiting to link up:', stream_id);
+                unlinkedInputs.set(stream_id, {options});
+                on_ok(stream_id);
+                return;
+            }
+
             if (engine.addInput(owner, stream_id, codec, conn)) {
                 inputs[stream_id] = {owner: owner,
                                      connection: conn};
@@ -261,7 +272,12 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
 
     that.unpublish = function (stream_id) {
         log.debug('unpublish, stream_id:', stream_id);
-        removeInput(stream_id);
+        if (fromMap.has(stream_id)) {
+            removeInput(fromMap.get(stream_id));
+            fromMap.delete(stream_id);
+        } else {
+            removeInput(stream_id);
+        }
     };
 
     that.setInputActive = function (stream_id, active, callback) {
@@ -283,13 +299,41 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         log.debug('unsubscribe, connectionId:', connectionId);
     };
 
-    that.linkup = function (connectionId, audio_stream_id, callback) {
-        log.debug('linkup, connectionId:', connectionId, 'audio_stream_id:', audio_stream_id);
+    // streamInfo = {id: 'string', ip: 'string', port: 'number'}
+    // from = {audio: streamInfo, video: streamInfo, data: streamInfo}
+    that.linkup = function (connectionId, from, callback) {
+        log.debug('linkup, connectionId:', connectionId, 'from:', from);
+        if (unlinkedInputs.has(connectionId)) {
+            const {options} = unlinkedInputs.get(connectionId);
+            unlinkedInputs.delete(connectionId);
+            const fromId = from.video.id;
+            options.ip = from.video?.ip;
+            options.port = from.video?.port;
+            addInput(fromId, options.publisher, options.audio.codec, options, function () {
+                if (unlinkedInputs.has(fromId)) {
+                    // Inputs link failed
+                    unlinkedInputs.delete(fromId);
+                    callback('callback', 'error', 'Invalid link address');
+                } else {
+                    fromMap.set(connectionId, fromId);
+                    callback('callback', 'ok');
+                }
+            }, function (reason) {
+                log.error(reason);
+                callback('callback', 'error', reason);
+            });
+            return;
+        }
         callback('callback', 'ok');
     };
 
     that.cutoff = function (connectionId) {
         log.debug('cutoff, connectionId:', connectionId);
+        if (fromMap.has(connectionId)) {
+            const fromId = fromMap.get(connectionId);
+            fromMap.delete(connectionId);
+            connectionId = fromId;
+        }
         if (connections[connectionId] && connections[connectionId].audioFrom) {
             if (outputs[connections[connectionId].audioFrom]) {
                 outputs[connections[connectionId].audioFrom].dispatcher.removeDestination('audio', connections[connectionId].connection.receiver());