Resolve comments

starwarfan · starwarfan · commit acbb773f5b4e · 2021-05-31T16:38:02.000+08:00
diff --git a/doc/design/owt-internal-transport.md b/doc/design/owt-internal-transport.md
@@ -2,7 +2,7 @@ Internal Transport
 
 # 1 Overview
 
-Internal transport use client-server model.
+Internal transport uses client-server model.
 
 An internal server for each process and one server can host multiple streams.
 
diff --git a/source/agent/addons/internalIO/InternalClientWrapper.cc b/source/agent/addons/internalIO/InternalClientWrapper.cc
@@ -22,10 +22,11 @@ InternalClient::InternalClient() {
 
 static void destroyAsyncHandle(uv_handle_t *handle) {
   delete handle;
+  handle = nullptr;
 }
 
 InternalClient::~InternalClient() {
-  boost::mutex::scoped_lock lock(mutex);
+  boost::mutex::scoped_lock lock(stats_lock);
   if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(async_stats_))) {
     ELOG_DEBUG("Closing Stats handle");
     uv_close(reinterpret_cast<uv_handle_t*>(async_stats_), destroyAsyncHandle);
@@ -132,7 +133,7 @@ NAUV_WORK_CB(InternalClient::statsCallback) {
     return;
   }
 
-  boost::mutex::scoped_lock lock(obj->mutex);
+  boost::mutex::scoped_lock lock(obj->stats_lock);
   while (!obj->stats_messages.empty()) {
     Local<Value> args[] = {
       Nan::New(obj->stats_messages.front().c_str()).ToLocalChecked()
@@ -146,7 +147,7 @@ NAUV_WORK_CB(InternalClient::statsCallback) {
 }
 
 void InternalClient::onConnected() {
-  boost::mutex::scoped_lock lock(mutex);
+  boost::mutex::scoped_lock lock(stats_lock);
   if (!async_stats_ || !stats_callback_) {
     return;
   }
@@ -156,7 +157,7 @@ void InternalClient::onConnected() {
 }
 
 void InternalClient::onDisconnected() {
-  boost::mutex::scoped_lock lock(mutex);
+  boost::mutex::scoped_lock lock(stats_lock);
   if (!async_stats_ || !stats_callback_) {
     return;
   }
diff --git a/source/agent/addons/internalIO/InternalClientWrapper.h b/source/agent/addons/internalIO/InternalClientWrapper.h
@@ -20,7 +20,7 @@ class InternalClient : public FrameSource,
     static NAN_MODULE_INIT(Init);
 
     owt_base::InternalClient* me;
-    boost::mutex mutex;
+    boost::mutex stats_lock;
     std::queue<std::string> stats_messages;
 
     // Implements owt_base::InternalClient::Listener
diff --git a/source/agent/addons/internalIO/InternalServerWrapper.cc b/source/agent/addons/internalIO/InternalServerWrapper.cc
@@ -22,10 +22,11 @@ InternalServer::InternalServer() {
 
 static void destroyAsyncHandle(uv_handle_t *handle) {
   delete handle;
+  handle = nullptr;
 }
 
 InternalServer::~InternalServer() {
-  boost::mutex::scoped_lock lock(mutex);
+  boost::mutex::scoped_lock lock(stats_lock);
   if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(async_stats_))) {
     ELOG_DEBUG("Closing Stats handle");
     uv_close(reinterpret_cast<uv_handle_t*>(async_stats_), destroyAsyncHandle);
@@ -65,7 +66,7 @@ NAN_METHOD(InternalServer::New) {
     InternalServer* obj = new InternalServer();
     obj->me = new owt_base::InternalServer(
         protocol, minPort, maxPort, obj);
-    if (info.Length() > 3) {
+    if (info.Length() > 3 && info[3]->IsFunction()) {
       obj->stats_callback_ = new Nan::Callback(info[3].As<Function>());
     }
     obj->Wrap(info.This());
@@ -122,7 +123,7 @@ NAUV_WORK_CB(InternalServer::statsCallback) {
   if (!obj || !obj->me || !obj->stats_callback_) {
     return;
   }
-  boost::mutex::scoped_lock lock(obj->mutex);
+  boost::mutex::scoped_lock lock(obj->stats_lock);
   while (!obj->stats_messages.empty()) {
     Local<Value> args[] = {
       Nan::New(obj->stats_messages.front().first.c_str()).ToLocalChecked(),
@@ -137,7 +138,7 @@ NAUV_WORK_CB(InternalServer::statsCallback) {
 }
 
 void InternalServer::onConnected(const std::string& id) {
-  boost::mutex::scoped_lock lock(mutex);
+  boost::mutex::scoped_lock lock(stats_lock);
   if (!async_stats_ || !stats_callback_) {
     return;
   }
@@ -147,7 +148,7 @@ void InternalServer::onConnected(const std::string& id) {
 }
 
 void InternalServer::onDisconnected(const std::string& id) {
-  boost::mutex::scoped_lock lock(mutex);
+  boost::mutex::scoped_lock lock(stats_lock);
   if (!async_stats_ || !stats_callback_) {
     return;
   }
diff --git a/source/agent/addons/internalIO/InternalServerWrapper.h b/source/agent/addons/internalIO/InternalServerWrapper.h
@@ -10,6 +10,9 @@
 #include <logger.h>
 #include <nan.h>
 
+#include <queue>
+#include <string>
+
 /*
  * Wrapper class of owt_base::InternalServer
  */
@@ -20,7 +23,7 @@ class InternalServer : public node::ObjectWrap,
     static NAN_MODULE_INIT(Init);
 
     owt_base::InternalServer* me;
-    boost::mutex mutex;
+    boost::mutex stats_lock;
     std::queue<std::pair<std::string, std::string>> stats_messages;
 
     // Implements owt_base::InternalServer::Listener
diff --git a/source/agent/audio/dist.json b/source/agent/audio/dist.json
@@ -30,7 +30,6 @@
                 "selectiveMixer.js",
                 "activeAudioSelector.js",
                 "../connections.js",
-                "../InternalConnectionFactory.js",
                 "../internalConnectionRouter.js"
             ]
         }
diff --git a/source/agent/audio/index.js b/source/agent/audio/index.js
@@ -16,7 +16,6 @@ var logger = require('../logger').logger;
 
 // Logger
 var log = logger.getLogger('AudioNode');
-var InternalConnectionFactory = require('./InternalConnectionFactory');
 var {InternalConnectionRouter} = require('./internalConnectionRouter');
 
 
@@ -51,10 +50,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
          */
         connections = {},
 
-        router = new InternalConnectionRouter(global.config.internal),
-
-        // For internal SCTP connection creation
-        internalConnFactory = new InternalConnectionFactory;
+        router = new InternalConnectionRouter(global.config.internal);
 
     var addInput = function (stream_id, owner, codec, options, on_ok, on_error) {
         if (engine) {
diff --git a/source/agent/recording/dist.json b/source/agent/recording/dist.json
@@ -28,7 +28,6 @@
             "recording": [
                 "index.js",
                 "../connections.js",
-                "../InternalConnectionFactory.js",
                 "../internalConnectionRouter.js"
             ]
         }
diff --git a/source/agent/recording/index.js b/source/agent/recording/index.js
@@ -18,7 +18,6 @@ var Connections = require('./connections');
 // Logger
 var log = logger.getLogger('RecordingNode');
 
-var InternalConnectionFactory = require('./InternalConnectionFactory');
 var {InternalConnectionRouter} = require('./internalConnectionRouter');
 
 
@@ -28,7 +27,6 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
       clusterIP: clusterWorkerIP
     };
     var connections = new Connections;
-    var internalConnFactory = new InternalConnectionFactory;
     var router = new InternalConnectionRouter(global.config.internal);
 
     var notifyStatus = (controller, sessionId, direction, status) => {
diff --git a/source/agent/sip/dist.json b/source/agent/sip/dist.json
@@ -27,8 +27,7 @@
                 "index.js",
                 "sipCallConnection.js",
                 "../connections.js",
-                "../internalConnectionRouter.js",
-                "../InternalConnectionFactory.js"
+                "../internalConnectionRouter.js"
             ]
         }
     },
diff --git a/source/agent/sip/index.js b/source/agent/sip/index.js
@@ -17,7 +17,6 @@ var cluster_name = ((global.config || {}).cluster || {}).name || 'owt-cluster';
 
 // Logger
 var log = logger.getLogger('SipNode');
-var InternalConnectionFactory = require('./InternalConnectionFactory');
 var {InternalConnectionRouter} = require('./internalConnectionRouter');
 
 // resolution map
@@ -235,8 +234,7 @@ module.exports = function (rpcC, selfRpcId, parentRpcId, clusterWorkerIP) {
         streams = {},
         calls = {},
         subscriptions = {},
-        recycling_mode = false,
-        internalConnFactory = new InternalConnectionFactory;
+        recycling_mode = false;
 
     var router = new InternalConnectionRouter(global.config.internal);
 
@@ -679,21 +677,6 @@ module.exports = function (rpcC, selfRpcId, parentRpcId, clusterWorkerIP) {
         recycling_mode = false;
     };
 
-    that.createInternalConnection = function (connectionId, direction, internalOpt, callback) {
-        // internalOpt.minport = global.config.internal.minport;
-        // internalOpt.maxport = global.config.internal.maxport;
-        var portInfo = router.internalPort;
-        // if (direction === 'in') {
-        //     internalConnFactory.create(connectionId, direction, internalOpt);
-        // }
-        callback('callback', {ip: global.config.internal.ip_address, 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;
diff --git a/source/agent/streaming/dist.json b/source/agent/streaming/dist.json
@@ -28,7 +28,6 @@
             "streaming": [
                 "index.js",
                 "../connections.js",
-                "../InternalConnectionFactory.js",
                 "../internalConnectionRouter.js"
             ]
         }
diff --git a/source/agent/streaming/index.js b/source/agent/streaming/index.js
@@ -18,7 +18,6 @@ var Connections = require('./connections');
 // Logger
 var log = logger.getLogger('StreamingNode');
 
-var InternalConnectionFactory = require('./InternalConnectionFactory');
 var {InternalConnectionRouter} = require('./internalConnectionRouter');
 
 module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
@@ -27,7 +26,6 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
       clusterIP: clusterWorkerIP
     };
     var connections = new Connections;
-    var internalConnFactory = new InternalConnectionFactory;
     var router = new InternalConnectionRouter(global.config.internal);
 
     var notifyStatus = (controller, sessionId, direction, status) => {
diff --git a/source/agent/video/dist.json b/source/agent/video/dist.json
@@ -37,7 +37,6 @@
                 "vmixer.js",
                 "vtranscoder.js",
                 "../connections.js",
-                "../InternalConnectionFactory.js",
                 "../internalConnectionRouter.js"
             ],
             "avatars": [
diff --git a/source/agent/webrtc/dist.json b/source/agent/webrtc/dist.json
@@ -33,7 +33,6 @@
                 "profileFilter.js",
                 "sdpInfo.js",
                 "../connections.js",
-                "../InternalConnectionFactory.js",
                 "../internalConnectionRouter.js"
             ],
             "cert": [
diff --git a/source/agent/webrtc/index.js b/source/agent/webrtc/index.js
@@ -7,7 +7,6 @@ var path = require('path');
 
 var WrtcConnection = require('./wrtcConnection');
 var Connections = require('./connections');
-var InternalConnectionFactory = require('./InternalConnectionFactory');
 var logger = require('../logger').logger;
 var {InternalConnectionRouter} = require('./internalConnectionRouter');
 
@@ -29,7 +28,6 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
       clusterIP: clusterWorkerIP
     };
     var connections = new Connections;
-    var internalConnFactory = new InternalConnectionFactory;
 
     var router = new InternalConnectionRouter(global.config.internal);
 
diff --git a/source/core/owt_base/internal/InternalClient.cpp b/source/core/owt_base/internal/InternalClient.cpp
@@ -1,4 +1,4 @@
-// Copyright (C) <2019> Intel Corporation
+// Copyright (C) <2021> Intel Corporation
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -52,12 +52,12 @@ void InternalClient::onFeedback(const FeedbackMsg& msg)
     }
     ELOG_DEBUG("onFeedback ");
 
-    char sendBuffer[512];
+    uint8_t sendBuffer[512];
     sendBuffer[0] = TDT_FEEDBACK_MSG;
     memcpy(&sendBuffer[1],
-           reinterpret_cast<char*>(const_cast<FeedbackMsg*>(&msg)),
+           reinterpret_cast<uint8_t*>(const_cast<FeedbackMsg*>(&msg)),
            sizeof(FeedbackMsg));
-    m_client->sendData((char*)sendBuffer, sizeof(FeedbackMsg) + 1);
+    m_client->sendData((uint8_t*)sendBuffer, sizeof(FeedbackMsg) + 1);
 }
 
 void InternalClient::onConnected()
@@ -80,11 +80,15 @@ void InternalClient::onConnected()
     m_ready = true;
 }
 
-void InternalClient::onData(char* buf, int len)
+void InternalClient::onData(uint8_t* buf, uint32_t len)
 {
     Frame* frame = nullptr;
     MetaData* metadata = nullptr;
-    switch (buf[0]) {
+    if (len <= 1) {
+        ELOG_DEBUG("Skip onData len: %u", (unsigned int)len);
+        return;
+    }
+    switch ((char) buf[0]) {
         case TDT_MEDIA_FRAME:
             frame = reinterpret_cast<Frame*>(buf + 1);
             frame->payload = reinterpret_cast<uint8_t*>(buf + 1 + sizeof(Frame));
diff --git a/source/core/owt_base/internal/InternalClient.h b/source/core/owt_base/internal/InternalClient.h
@@ -40,7 +40,7 @@ class InternalClient : public FrameSource,
 
     // Implements TransportClient::Listener
     void onConnected() override;
-    void onData(char* data, int len) override;
+    void onData(uint8_t* data, uint32_t len) override;
     void onDisconnected() override;
 
 private:
diff --git a/source/core/owt_base/internal/InternalServer.cpp b/source/core/owt_base/internal/InternalServer.cpp
@@ -1,4 +1,4 @@
-// Copyright (C) <2019> Intel Corporation
+// Copyright (C) <2021> Intel Corporation
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -81,7 +81,7 @@ void InternalServer::onSessionAdded(int id)
     }
 }
 
-void InternalServer::onSessionData(int id, char* data, int len)
+void InternalServer::onSessionData(int id, uint8_t* data, uint32_t len)
 {
     if (!m_sessions.count(id)) {
         ELOG_WARN("Unknown ID:%d for onSessionData", id);
@@ -158,10 +158,10 @@ void InternalServer::onSessionRemoved(int id)
 
 void InternalServer::InternalSession::onFrame(const Frame& frame)
 {
-    char sendBuffer[1 + sizeof(Frame) + frame.length];
+    uint8_t sendBuffer[1 + sizeof(Frame) + frame.length];
     sendBuffer[0] = TDT_MEDIA_FRAME;
     memcpy(&sendBuffer[1],
-           reinterpret_cast<char*>(const_cast<Frame*>(&frame)),
+           reinterpret_cast<uint8_t*>(const_cast<Frame*>(&frame)),
            sizeof(Frame));
     memcpy(&sendBuffer[1 + sizeof(Frame)], frame.payload, frame.length);
 
@@ -170,10 +170,10 @@ void InternalServer::InternalSession::onFrame(const Frame& frame)
 
 void InternalServer::InternalSession::onMetaData(const MetaData& metadata)
 {
-    char sendBuffer[1 + sizeof(MetaData) + metadata.length];
+    uint8_t sendBuffer[1 + sizeof(MetaData) + metadata.length];
     sendBuffer[0] = TDT_MEDIA_METADATA;
     memcpy(&sendBuffer[1],
-           reinterpret_cast<char*>(const_cast<MetaData*>(&metadata)),
+           reinterpret_cast<uint8_t*>(const_cast<MetaData*>(&metadata)),
            sizeof(MetaData));
     memcpy(&sendBuffer[1 + sizeof(MetaData)], metadata.payload, metadata.length);
 
diff --git a/source/core/owt_base/internal/InternalServer.h b/source/core/owt_base/internal/InternalServer.h
@@ -35,7 +35,7 @@ class InternalServer : public TransportServer::Listener {
 
     // Implements TransportServer::Listener
     void onSessionAdded(int id) override;
-    void onSessionData(int id, char* data, int len) override;
+    void onSessionData(int id, uint8_t* data, uint32_t len) override;
     void onSessionRemoved(int id) override;
 
 private:
diff --git a/source/core/owt_base/internal/TransportBase.cpp b/source/core/owt_base/internal/TransportBase.cpp
diff --git a/source/core/owt_base/internal/TransportBase.h b/source/core/owt_base/internal/TransportBase.h
diff --git a/source/core/owt_base/internal/TransportClient.cpp b/source/core/owt_base/internal/TransportClient.cpp
diff --git a/source/core/owt_base/internal/TransportClient.h b/source/core/owt_base/internal/TransportClient.h
diff --git a/source/core/owt_base/internal/TransportServer.cpp b/source/core/owt_base/internal/TransportServer.cpp
diff --git a/source/core/owt_base/internal/TransportServer.h b/source/core/owt_base/internal/TransportServer.h

Original file line number	Diff line number	Diff line change
`@@ -22,10 +22,11 @@ InternalClient::InternalClient() {`
`22`	`22`
`23`	`23`	`static void destroyAsyncHandle(uv_handle_t *handle) {`
`24`	`24`	`delete handle;`
	`25`	`+ handle = nullptr;`
`25`	`26`	`}`
`26`	`27`
`27`	`28`	`InternalClient::~InternalClient() {`
`28`		`- boost::mutex::scoped_lock lock(mutex);`
	`29`	`+ boost::mutex::scoped_lock lock(stats_lock);`
`29`	`30`	`if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(async_stats_))) {`
`30`	`31`	`ELOG_DEBUG("Closing Stats handle");`
`31`	`32`	`uv_close(reinterpret_cast<uv_handle_t*>(async_stats_), destroyAsyncHandle);`
`@@ -132,7 +133,7 @@ NAUV_WORK_CB(InternalClient::statsCallback) {`
`132`	`133`	`return;`
`133`	`134`	`}`
`134`	`135`
`135`		`- boost::mutex::scoped_lock lock(obj->mutex);`
	`136`	`+ boost::mutex::scoped_lock lock(obj->stats_lock);`
`136`	`137`	`while (!obj->stats_messages.empty()) {`
`137`	`138`	`Local<Value> args[] = {`
`138`	`139`	`Nan::New(obj->stats_messages.front().c_str()).ToLocalChecked()`
`@@ -146,7 +147,7 @@ NAUV_WORK_CB(InternalClient::statsCallback) {`
`146`	`147`	`}`
`147`	`148`
`148`	`149`	`void InternalClient::onConnected() {`
`149`		`- boost::mutex::scoped_lock lock(mutex);`
	`150`	`+ boost::mutex::scoped_lock lock(stats_lock);`
`150`	`151`	`if (!async_stats_ \|\| !stats_callback_) {`
`151`	`152`	`return;`
`152`	`153`	`}`
`@@ -156,7 +157,7 @@ void InternalClient::onConnected() {`
`156`	`157`	`}`
`157`	`158`
`158`	`159`	`void InternalClient::onDisconnected() {`
`159`		`- boost::mutex::scoped_lock lock(mutex);`
	`160`	`+ boost::mutex::scoped_lock lock(stats_lock);`
`160`	`161`	`if (!async_stats_ \|\| !stats_callback_) {`
`161`	`162`	`return;`
`162`	`163`	`}`
Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,6 @@`
`30`	`30`	`"selectiveMixer.js",`
`31`	`31`	`"activeAudioSelector.js",`
`32`	`32`	`"../connections.js",`
`33`		`- "../InternalConnectionFactory.js",`
`34`	`33`	`"../internalConnectionRouter.js"`
`35`	`34`	`]`
`36`	`35`	`}`
Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,6 @@`
`28`	`28`	`"recording": [`
`29`	`29`	`"index.js",`
`30`	`30`	`"../connections.js",`
`31`		`- "../InternalConnectionFactory.js",`
`32`	`31`	`"../internalConnectionRouter.js"`
`33`	`32`	`]`
`34`	`33`	`}`
Original file line number	Diff line number	Diff line change
`@@ -27,8 +27,7 @@`
`27`	`27`	`"index.js",`
`28`	`28`	`"sipCallConnection.js",`
`29`	`29`	`"../connections.js",`
`30`		`- "../internalConnectionRouter.js",`
`31`		`- "../InternalConnectionFactory.js"`
	`30`	`+ "../internalConnectionRouter.js"`
`32`	`31`	`]`
`33`	`32`	`}`
`34`	`33`	`},`