Use tickets for internal connection (#932)

starwarfan · web-flow · commit 7c5ccf5287b9 · 2021-03-17T15:07:43.000+08:00
* Add connect ticket for RawTransport
diff --git a/source/agent/InternalConnectionFactory.js b/source/agent/InternalConnectionFactory.js
@@ -39,15 +39,15 @@ try {
 }
 
 // Wrapper object for sctp-connection and tcp/udp-connection
-function InConnection(prot, minport, maxport) {
+function InConnection(prot, minport, maxport, ticket) {
     var conn = null;
     var protocol = "quic";
 
     switch (prot) {
         case 'tcp':
         case 'udp':
             protocol = prot;
-            conn = new InternalIn(prot, minport, maxport);
+            conn = new InternalIn(prot, minport, maxport, ticket);
             break;
         case 'quic':
             conn = new quicIO.in(cf, kf);
@@ -78,7 +78,7 @@ function InConnection(prot, minport, maxport) {
 }
 
 // Wrapper object for sctp-connection and tcp/udp-connection
-function OutConnection(prot, minport, maxport) {
+function OutConnection(prot, minport, maxport, ticket) {
     var that = {};
     var conn = null;
     var protocol = "quic";
@@ -111,7 +111,7 @@ function OutConnection(prot, minport, maxport) {
         } else if (protocol === 'quic') {
             conn = new quicIO.out(connectOpt.ip, connectOpt.port);
         } else {
-            conn = new InternalOut(protocol, connectOpt.ip, connectOpt.port);
+            conn = new InternalOut(protocol, connectOpt.ip, connectOpt.port, ticket);
         }
     };
 
@@ -139,6 +139,7 @@ module.exports = function() {
         var prot = internalOpt.protocol;
         var minport = internalOpt.minport || 0;
         var maxport = internalOpt.maxport || 0;
+        var ticket = internalOpt.ticket;
 
         if (preparedSet[connId]) {
             log.warn('Internal Connection already prepared:', connId);
@@ -147,7 +148,9 @@ module.exports = function() {
             // right call sequence in upper layer.
             return preparedSet[connId].connection.getListeningPort();
         }
-        var conn = (direction === 'in')? InConnection(prot, minport, maxport) : OutConnection(prot, minport, maxport);
+        var conn = (direction === 'in')
+            ? InConnection(prot, minport, maxport, ticket)
+            : OutConnection(prot, minport, maxport, ticket);
 
         preparedSet[connId] = {connection: conn, direction: direction};
         return conn.getListeningPort();
diff --git a/source/agent/addons/internalIO/InternalInWrapper.cc b/source/agent/addons/internalIO/InternalInWrapper.cc
@@ -37,14 +37,23 @@ void InternalIn::New(const FunctionCallbackInfo<Value>& args) {
   String::Utf8Value param0(args[0]->ToString());
   std::string protocol = std::string(*param0);
   unsigned int minPort = 0, maxPort = 0;
+  std::string ticket;
 
   if (args.Length() >= 3) {
     minPort = args[1]->Uint32Value();
     maxPort = args[2]->Uint32Value();
   }
+  if (args.Length() > 3) {
+    String::Utf8Value param3(args[3]->ToString());
+    ticket = std::string(*param3);
+  }
 
   InternalIn* obj = new InternalIn();
-  obj->me = new owt_base::InternalIn(protocol, minPort, maxPort);
+  if (ticket.empty()) {
+    obj->me = new owt_base::InternalIn(protocol, minPort, maxPort);
+  } else {
+    obj->me = new owt_base::InternalIn(protocol, ticket, minPort, maxPort);
+  }
   obj->src = obj->me;
 
   obj->Wrap(args.This());
diff --git a/source/agent/addons/internalIO/InternalOutWrapper.cc b/source/agent/addons/internalIO/InternalOutWrapper.cc
@@ -36,9 +36,18 @@ void InternalOut::New(const v8::FunctionCallbackInfo<v8::Value>& args) {
   String::Utf8Value param1(args[1]->ToString());
   std::string dest_ip = std::string(*param1);
   unsigned int dest_port = args[2]->Uint32Value();
+  std::string ticket;
+  if (args.Length() > 3) {
+    String::Utf8Value param3(args[3]->ToString());
+    ticket = std::string(*param3);
+  }
 
   InternalOut* obj = new InternalOut();
-  obj->me = new owt_base::InternalOut(protocol, dest_ip, dest_port);
+  if (ticket.empty()) {
+    obj->me = new owt_base::InternalOut(protocol, dest_ip, dest_port);
+  } else {
+    obj->me = new owt_base::InternalOut(protocol, ticket, dest_ip, dest_port);
+  }
   obj->dest = obj->me;
 
   obj->Wrap(args.This());
diff --git a/source/agent/analytics/analytics-agent.js b/source/agent/analytics/analytics-agent.js
@@ -25,14 +25,16 @@ class AnalyticsAgent extends BaseAgent {
     this.engine = new VideoAnalyzer();
 
     this.flag = 0;
+    this.ticket = null;
   }
 
 // override
 createInternalConnection(connectionId, direction, internalOpt) {
     internalOpt.minport = global.config.internal.minport;
     internalOpt.maxport = global.config.internal.maxport;
+    this.ticket = internalOpt.ticket;
     if (direction == 'in') {
-      this.engine.emitListenTo(internalOpt.minport,internalOpt.maxport);
+      this.engine.emitListenTo(internalOpt.minport,internalOpt.maxport, this.ticket);
       const portInfo = this.engine.getListeningPort();
       // Create internal connection always success
       return Promise.resolve({ip: global.config.internal.ip_address, port: portInfo});
diff --git a/source/agent/analytics/videoGstPipeline/GstInternalIn.cpp b/source/agent/analytics/videoGstPipeline/GstInternalIn.cpp
@@ -19,10 +19,11 @@ static void dump(void* index, uint8_t* buf, int len)
 }
 
 DEFINE_LOGGER(GstInternalIn, "GstInternalIn");
-GstInternalIn::GstInternalIn(GstAppSrc *data, unsigned int minPort, unsigned int maxPort)
+GstInternalIn::GstInternalIn(GstAppSrc *data, unsigned int minPort, unsigned int maxPort, std::string ticket)
 {
     m_transport.reset(new owt_base::RawTransport<owt_base::TCP>(this));
 
+    m_transport->initTicket(ticket);
     if (minPort > 0 && minPort <= maxPort) {
         m_transport->listenTo(minPort, maxPort);
     } else {
diff --git a/source/agent/analytics/videoGstPipeline/GstInternalIn.h b/source/agent/analytics/videoGstPipeline/GstInternalIn.h
@@ -16,7 +16,7 @@
 class GstInternalIn : public owt_base::RawTransportListener{
     DECLARE_LOGGER();
 public:
-    GstInternalIn(GstAppSrc *data, unsigned int minPort = 0, unsigned int maxPort = 0);
+    GstInternalIn(GstAppSrc *data, unsigned int minPort = 0, unsigned int maxPort = 0, std::string ticket = NULL);
     virtual ~GstInternalIn();
 
     unsigned int getListeningPort();
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.cc b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.cc
@@ -352,10 +352,9 @@ int VideoGstAnalyzer::setPlaying()
     return 0;
 }
 
-void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort)
-{
+void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort, std::string ticket) {
     ELOG_DEBUG("Listening\n");
-    m_internalin.reset(new GstInternalIn((GstAppSrc*)source, minPort, maxPort));  
+    m_internalin.reset(new GstInternalIn((GstAppSrc*)source, minPort, maxPort, ticket));
 }
 
 void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* out)
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.h b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.h
@@ -30,7 +30,7 @@ class VideoGstAnalyzer : public EventRegistry {
     void clearPipeline();
     void destroyPipeline();
     int getListeningPort();
-    void emitListenTo(int minPort,int maxPort);
+    void emitListenTo(int minPort, int maxPort, std::string ticket);
     int setPlaying();
 
     int addElementMany();
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzerWrap.cc b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzerWrap.cc
@@ -97,11 +97,17 @@ void VideoGstAnalyzerWrap::emitListenTo(const FunctionCallbackInfo<Value>& args)
   mcu::VideoGstAnalyzer* me = obj->me;
 
   unsigned int minPort = 0, maxPort = 0;
+  std::string ticket;
 
+  if (args.Length() >= 3) {
     minPort = args[0]->Uint32Value();
     maxPort = args[1]->Uint32Value();
 
-  me->emitListenTo(minPort,maxPort);
+    String::Utf8Value param3(args[2]->ToString());
+    ticket = std::string(*param3);
+  }
+  me->emitListenTo(minPort, maxPort, ticket);
+
 }
 
 
diff --git a/source/agent/conference/package.json b/source/agent/conference/package.json
@@ -7,6 +7,7 @@
     "node-getopt": "*",
     "toml": "*",
     "mongoose": "^5.9.6",
+    "uuid": "^8.0.0",
     "fraction.js": "^4.0.12"
   },
   "devDependencies": {
diff --git a/source/agent/conference/roomController.js b/source/agent/conference/roomController.js
@@ -11,6 +11,8 @@ var makeRPC = require('./makeRPC').makeRPC;
 // Logger
 var log = logger.getLogger('RoomController');
 
+const { v4 : uuid } = require('uuid');
+
 const {
     isVideoFmtCompatible,
     isResolutionEqual,
@@ -40,12 +42,14 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         rpcClient = spec.rpcClient,
         config = spec.config,
         room_id = spec.room,
-	origin = spec.origin,
+        origin = spec.origin,
         selfRpcId = spec.selfRpcId,
         enable_audio_transcoding = config.transcoding && !!config.transcoding.audio,
         enable_video_transcoding = config.transcoding && !!config.transcoding.video,
         internal_conn_protocol = config.internalConnProtocol;
 
+    var internalTicket = uuid().replace(/-/g, '');
+
     /*
     mix_views = {
         view: {
@@ -524,7 +528,8 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
 
         // Transport protocol for creating internal connection
         var internalOpt = {
-            protocol: internal_conn_protocol
+            protocol: internal_conn_protocol,
+            ticket: internalTicket,
         };
         var from, to, has_published, has_subscribed;
 
diff --git a/source/core/owt_base/InternalIn.cpp b/source/core/owt_base/InternalIn.cpp
@@ -20,6 +20,25 @@ InternalIn::InternalIn(const std::string& protocol, unsigned int minPort, unsign
     }
 }
 
+InternalIn::InternalIn(
+    const std::string& protocol,
+    const std::string& ticket,
+    unsigned int minPort,
+    unsigned int maxPort)
+{
+    if (protocol == "tcp")
+        m_transport.reset(new owt_base::RawTransport<TCP>(this));
+    else
+        m_transport.reset(new owt_base::RawTransport<UDP>(this, 64 * 1024));
+
+    m_transport->initTicket(ticket);
+    if (minPort > 0 && minPort <= maxPort) {
+        m_transport->listenTo(minPort, maxPort);
+    } else {
+        m_transport->listenTo(0);
+    }
+}
+
 InternalIn::~InternalIn()
 {
     m_transport->close();
diff --git a/source/core/owt_base/InternalIn.h b/source/core/owt_base/InternalIn.h
@@ -13,6 +13,8 @@ namespace owt_base {
 class InternalIn : public FrameSource, public RawTransportListener {
 public:
     InternalIn(const std::string& protocol, unsigned int minPort = 0, unsigned int maxPort = 0);
+    InternalIn(const std::string& protocol, const std::string& ticket,
+               unsigned int minPort = 0, unsigned int maxPort = 0);
     virtual ~InternalIn();
 
     unsigned int getListeningPort();
diff --git a/source/core/owt_base/InternalOut.cpp b/source/core/owt_base/InternalOut.cpp
@@ -16,6 +16,21 @@ InternalOut::InternalOut(const std::string& protocol, const std::string& dest_ip
     m_transport->createConnection(dest_ip, dest_port);
 }
 
+InternalOut::InternalOut(
+    const std::string& protocol,
+    const std::string& ticket,
+    const std::string& dest_ip,
+    unsigned int dest_port)
+{
+    if (protocol == "tcp")
+        m_transport.reset(new owt_base::RawTransport<TCP>(this));
+    else
+        m_transport.reset(new owt_base::RawTransport<UDP>(this));
+
+    m_transport->initTicket(ticket);
+    m_transport->createConnection(dest_ip, dest_port);
+}
+
 InternalOut::~InternalOut()
 {
     m_transport->close();
@@ -51,6 +66,5 @@ void InternalOut::onTransportData(char* buf, int len)
     }
 }
 
-
 } /* namespace owt_base */
 
diff --git a/source/core/owt_base/InternalOut.h b/source/core/owt_base/InternalOut.h
@@ -13,6 +13,8 @@ namespace owt_base {
 class InternalOut : public FrameDestination, public RawTransportListener {
 public:
     InternalOut(const std::string& protocol, const std::string& dest_ip, unsigned int dest_port);
+    InternalOut(const std::string& protocol, const std::string& ticket,
+                const std::string& dest_ip, unsigned int dest_port);
     virtual ~InternalOut();
 
     void onFrame(const Frame&);
diff --git a/source/core/owt_base/RawTransport.cpp b/source/core/owt_base/RawTransport.cpp
diff --git a/source/core/owt_base/RawTransport.h b/source/core/owt_base/RawTransport.h

Original file line number	Diff line number	Diff line change
`@@ -19,10 +19,11 @@ static void dump(void* index, uint8_t* buf, int len)`
`19`	`19`	`}`
`20`	`20`
`21`	`21`	`DEFINE_LOGGER(GstInternalIn, "GstInternalIn");`
`22`		`-GstInternalIn::GstInternalIn(GstAppSrc *data, unsigned int minPort, unsigned int maxPort)`
	`22`	`+GstInternalIn::GstInternalIn(GstAppSrc *data, unsigned int minPort, unsigned int maxPort, std::string ticket)`
`23`	`23`	`{`
`24`	`24`	`m_transport.reset(new owt_base::RawTransport<owt_base::TCP>(this));`
`25`	`25`
	`26`	`+ m_transport->initTicket(ticket);`
`26`	`27`	`if (minPort > 0 && minPort <= maxPort) {`
`27`	`28`	`m_transport->listenTo(minPort, maxPort);`
`28`	`29`	`} else {`
Original file line number	Diff line number	Diff line change
`@@ -352,10 +352,9 @@ int VideoGstAnalyzer::setPlaying()`
`352`	`352`	`return 0;`
`353`	`353`	`}`
`354`	`354`
`355`		`-void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort)`
`356`		`-{`
	`355`	`+void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort, std::string ticket) {`
`357`	`356`	`ELOG_DEBUG("Listening\n");`
`358`		`- m_internalin.reset(new GstInternalIn((GstAppSrc*)source, minPort, maxPort));`
	`357`	`+ m_internalin.reset(new GstInternalIn((GstAppSrc*)source, minPort, maxPort, ticket));`
`359`	`358`	`}`
`360`	`359`
`361`	`360`	`void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* out)`