Merge pull request #988 from starwarfan/internal-rework

Refactor internal transport for reducing listening ports

Author: starwarfan

doc/design/owt-internal-transport.md

@@ -0,0 +1,89 @@
+Internal Transport
+
+# 1 Overview
+
+Internal transport uses client-server model.
+
+An internal server for each process and one server can host multiple streams.
+
+An internal client for each remote stream, client connects server to receive stream data.
+
+The internal connection can be TCP/TLS/QUIC.
+
+![plot](./pics/internal_structure.png)
+
+# 2 Workflow
+
+![plot](./pics/internal_connect.png)
+
+(Certificate to verify client)
+
+Drop data policy:
+
+1.  Drop non-key frame data for video.
+
+2.  Drop low level data for audio.
+
+3.  Drop oldest data.
+
+4.  Other customized policy based on media information.
+
+# 3 Data format
+
+The transport message over internal connection is in following format:
+
+| 4 bytes (payload length K) | K bytes (payload data) |
+
+Payload data = | 1 byte (message type) | (k – 1 ) bytes format data |
+
+The “message type” can be: (protobuf)
+
+1.  Feedback (client to server)
+
+2.  MediaFrame (server to client)
+
+3.  MetaData (server to client)
+
+The payload format of each type can be referred from MediaFramePipeline.cpp.
+
+Feedback – class FeedbackMsg
+
+MediaFrame – class Frame
+
+MetaData – class MetaData
+
+# 4 Control layer
+
+## 4.1 Spread call
+
+Every stream has an internal address of server (ip, port).
+
+For each agent process, there’re local and remote streams like client
+SDK.
+
+`Local stream`: stream which is created on local process.
+
+`Remote stream`: stream which is spread from other nodes through
+internal connection.
+
+If we want to spread a stream to local, only internal address (ip, port) of that stream is needed.
+
+RPC calls for spreading:
+
+Conference node –(linkup with stream address)–> Target node (do spread and linkup)
+
+Linkup call:
+RPC linkup(
+    targetId, /* linkup destination, like webrtc subscription Id */
+    from = { /* linkup source */
+        type /* audio, video... */: {id, ip, port} /* remote stream description */
+    },
+)
+
+
+## 4.2 Lifetime
+
+A `TransportServer` lives during the `workingNode` process.
+
+A `TransportClient` is created for a `Remote stream`, is destroyed
+by disconnect event or close call.

doc/design/pics/internal_connect.png

@@ -0,0 +1,167 @@
+// Copyright (C) <2021> Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef BUILDING_NODE_EXTENSION
+#define BUILDING_NODE_EXTENSION
+#endif
+
+#include "InternalClientWrapper.h"
+
+using namespace v8;
+
+DEFINE_LOGGER(InternalClient, "InternalClientWrapper");
+
+Nan::Persistent<Function> InternalClient::constructor;
+
+InternalClient::InternalClient() {
+  async_stats_ = new uv_async_t;
+  stats_callback_ = nullptr;
+  uv_async_init(uv_default_loop(), async_stats_, &InternalClient::statsCallback);
+}
+
+static void destroyAsyncHandle(uv_handle_t *handle) {
+  delete handle;
+  handle = nullptr;
+}
+
+InternalClient::~InternalClient() {
+  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);
+  }
+  async_stats_ = nullptr;
+}
+
+NAN_MODULE_INIT(InternalClient::Init) {
+  // Constructor template
+  Local<FunctionTemplate> tpl = Nan::New<FunctionTemplate>(New);
+  tpl->SetClassName(Nan::New("InternalClient").ToLocalChecked());
+  tpl->InstanceTemplate()->SetInternalFieldCount(1);
+
+  // Prototype
+  Nan::SetPrototypeMethod(tpl, "close", close);
+  Nan::SetPrototypeMethod(tpl, "addDestination", addDestination);
+  Nan::SetPrototypeMethod(tpl, "removeDestination", removeDestination);
+
+  constructor.Reset(Nan::GetFunction(tpl).ToLocalChecked());
+  Nan::Set(target, Nan::New("InternalClient").ToLocalChecked(),
+           Nan::GetFunction(tpl).ToLocalChecked());
+}
+
+NAN_METHOD(InternalClient::New) {
+  if (info.Length() < 4) {
+    Nan::ThrowError("Wrong number of arguments");
+  }
+
+  if (info.IsConstructCall()) {
+    Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
+    std::string streamId = std::string(*param0);
+
+    Nan::Utf8String param1(Nan::To<v8::String>(info[1]).ToLocalChecked());
+    std::string protocol = std::string(*param1);
+
+    Nan::Utf8String param2(Nan::To<v8::String>(info[2]).ToLocalChecked());
+    std::string ip = std::string(*param2);
+
+    unsigned int port = Nan::To<unsigned int>(info[3]).FromJust();
+
+    InternalClient* obj = new InternalClient();
+    obj->me = new owt_base::InternalClient(
+        streamId, protocol, ip, port, obj);
+    if (info.Length() > 4 && info[4]->IsFunction()) {
+      obj->stats_callback_ = new Nan::Callback(info[4].As<Function>());
+    }
+    obj->src = obj->me;
+    obj->Wrap(info.This());
+    info.GetReturnValue().Set(info.This());
+  } else {
+    ELOG_WARN("Not construct call");
+  }
+}
+
+NAN_METHOD(InternalClient::close) {
+  InternalClient* obj = ObjectWrap::Unwrap<InternalClient>(info.Holder());
+  obj->src = nullptr;
+  delete obj->me;
+  obj->me = nullptr;
+}
+
+NAN_METHOD(InternalClient::addDestination) {
+  InternalClient* obj = ObjectWrap::Unwrap<InternalClient>(info.Holder());
+  owt_base::InternalClient* me = obj->me;
+
+  Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
+  std::string track = std::string(*param0);
+
+  FrameDestination* param =
+    ObjectWrap::Unwrap<FrameDestination>(
+      info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
+  owt_base::FrameDestination* dest = param->dest;
+
+  if (track == "audio") {
+    me->addAudioDestination(dest);
+  } else if (track == "video") {
+    me->addVideoDestination(dest);
+  }
+}
+
+NAN_METHOD(InternalClient::removeDestination) {
+  InternalClient* obj = ObjectWrap::Unwrap<InternalClient>(info.Holder());
+  owt_base::InternalClient* me = obj->me;
+
+  Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
+  std::string track = std::string(*param0);
+
+  FrameDestination* param =
+    ObjectWrap::Unwrap<FrameDestination>(
+      info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
+  owt_base::FrameDestination* dest = param->dest;
+
+  if (track == "audio") {
+    me->removeAudioDestination(dest);
+  } else if (track == "video") {
+    me->removeVideoDestination(dest);
+  }
+}
+
+NAUV_WORK_CB(InternalClient::statsCallback) {
+  Nan::HandleScope scope;
+  InternalClient* obj = reinterpret_cast<InternalClient*>(async->data);
+  if (!obj || !obj->me || !obj->stats_callback_) {
+    return;
+  }
+
+  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()
+    };
+    Nan::AsyncResource resource("InternalServer.statsCallback");
+    resource.runInAsyncScope(Nan::GetCurrentContext()->Global(),
+                             obj->stats_callback_->GetFunction(),
+                             1, args);
+    obj->stats_messages.pop();
+  }
+}
+
+void InternalClient::onConnected() {
+  boost::mutex::scoped_lock lock(stats_lock);
+  if (!async_stats_ || !stats_callback_) {
+    return;
+  }
+  stats_messages.push("connected");
+  async_stats_->data = this;
+  uv_async_send(async_stats_);
+}
+
+void InternalClient::onDisconnected() {
+  boost::mutex::scoped_lock lock(stats_lock);
+  if (!async_stats_ || !stats_callback_) {
+    return;
+  }
+  stats_messages.push("disconnected");
+  async_stats_->data = this;
+  uv_async_send(async_stats_);
+}

doc/design/pics/internal_structure.png

@@ -0,0 +1,50 @@
+// Copyright (C) <2021> Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef INTERNALCLIENTWRAPPER_H
+#define INTERNALCLIENTWRAPPER_H
+
+#include "../../addons/common/MediaFramePipelineWrapper.h"
+#include <InternalClient.h>
+#include <logger.h>
+#include <nan.h>
+
+/*
+ * Wrapper class of owt_base::InternalClient
+ */
+class InternalClient : public FrameSource,
+                       public owt_base::InternalClient::Listener {
+public:
+    DECLARE_LOGGER();
+    static NAN_MODULE_INIT(Init);
+
+    owt_base::InternalClient* me;
+    boost::mutex stats_lock;
+    std::queue<std::string> stats_messages;
+
+    // Implements owt_base::InternalClient::Listener
+    void onConnected() override;
+    void onDisconnected() override;
+
+private:
+    InternalClient();
+    ~InternalClient();
+
+    Nan::Callback *stats_callback_;
+    uv_async_t *async_stats_;
+
+    static Nan::Persistent<v8::Function> constructor;
+
+    static NAN_METHOD(New);
+
+    static NAN_METHOD(close);
+
+    static NAN_METHOD(addDestination);
+
+    static NAN_METHOD(removeDestination);
+
+    static NAUV_WORK_CB(statsCallback);
+};
+
+#endif

source/agent/addons/internalIO/InternalClientWrapper.cc

@@ -7,14 +7,15 @@
 #endif
 
 #include "InternalConfig.h"
-#include <RawTransport.h>
+#include <TransportBase.h>
 
 using namespace v8;
 
 void setPassphrase(const FunctionCallbackInfo<Value>& args) {
-  String::Utf8Value param0(args[0]->ToString());
+  Isolate* isolate = Isolate::GetCurrent();
+  String::Utf8Value param0(isolate, args[0]->ToString());
   std::string p = std::string(*param0);
-  owt_base::RawTransport<owt_base::Protocol::TCP>::setPassphrase(p);
+  owt_base::TransportSecret::setPassphrase(p);
 }
 
 void InitInternalConfig(v8::Local<v8::Object> exports) {