PlainTransportOptions: add optional rtcpListenInfo (#1099)

versatica · Jun 2, 2023 · 6594f07 · 6594f07
1 parent 64b4705
commit 6594f07
Show file tree

Hide file tree

Showing 9 changed files with 111 additions and 20 deletions.
diff --git a/node/src/PlainTransport.ts b/node/src/PlainTransport.ts
@@ -36,6 +36,11 @@ type PlainTransportListenInfo =
 	 * Listening info.
 	 */
 	listenInfo: TransportListenInfo;
+
+	/**
+	 * Optional listening info for RTCP.
+	 */
+	rtcpListenInfo?: TransportListenInfo;
 };
 
 type PlainTransportListenIp =

diff --git a/node/src/Router.ts b/node/src/Router.ts
@@ -615,6 +615,7 @@ export class Router<RouterAppData extends AppData = AppData>
 	async createPlainTransport<PlainTransportAppData extends AppData = AppData>(
 		{
 			listenInfo,
+			rtcpListenInfo,
 			listenIp,
 			port,
 			rtcpMux = true,
@@ -644,6 +645,14 @@ export class Router<RouterAppData extends AppData = AppData>
 			throw new TypeError('if given, appData must be an object');
 		}
 
+		// If rtcpMux is enabled, ignore rtcpListenInfo.
+		if (rtcpMux)
+		{
+			logger.warn('createPlainTransport() | ignoring given rtcpListenInfo since rtcpMux is enabled');
+
+			rtcpListenInfo = undefined;
+		}
+
 		// Convert deprecated TransportListenIps to TransportListenInfos.
 		if (listenIp)
 		{
@@ -688,6 +697,16 @@ export class Router<RouterAppData extends AppData = AppData>
 				listenInfo!.sendBufferSize,
 				listenInfo!.recvBufferSize
 			),
+			rtcpListenInfo ? new FbsTransport.ListenInfoT(
+				rtcpListenInfo.protocol === 'udp'
+					? FbsTransportProtocol.UDP
+					: FbsTransportProtocol.TCP,
+				rtcpListenInfo.ip,
+				rtcpListenInfo.announcedIp,
+				rtcpListenInfo.port,
+				rtcpListenInfo.sendBufferSize,
+				rtcpListenInfo.recvBufferSize
+			) : undefined,
 			rtcpMux,
 			comedia,
 			enableSrtp,

diff --git a/node/src/tests/test-PlainTransport.ts b/node/src/tests/test-PlainTransport.ts
@@ -119,22 +119,25 @@ test('router.createPlainTransport() succeeds', async () =>
 
 	expect(typeof anotherTransport).toBe('object');
 
+	const rtpPort = await pickPort({ ip: '127.0.0.1', reserveTimeout: 0 });
+	const rtcpPort = await pickPort({ ip: '127.0.0.1', reserveTimeout: 0 });
 	const transport2 = await router.createPlainTransport(
 		{
-			listenInfo : { protocol: 'udp', ip: '127.0.0.1' },
-			rtcpMux    : false
+			listenInfo     : { protocol: 'udp', ip: '127.0.0.1', port: rtpPort },
+			rtcpListenInfo : { protocol: 'udp', ip: '127.0.0.1', port: rtcpPort },
+			rtcpMux        : false
 		});
 
 	expect(typeof transport2.id).toBe('string');
 	expect(transport2.closed).toBe(false);
 	expect(transport2.appData).toEqual({});
 	expect(typeof transport2.tuple).toBe('object');
 	expect(transport2.tuple.localIp).toBe('127.0.0.1');
-	expect(typeof transport2.tuple.localPort).toBe('number');
+	expect(transport2.tuple.localPort).toBe(rtpPort);
 	expect(transport2.tuple.protocol).toBe('udp');
 	expect(typeof transport2.rtcpTuple).toBe('object');
 	expect(transport2.rtcpTuple?.localIp).toBe('127.0.0.1');
-	expect(typeof transport2.rtcpTuple?.localPort).toBe('number');
+	expect(transport2.rtcpTuple?.localPort).toBe(rtcpPort);
 	expect(transport2.rtcpTuple?.protocol).toBe('udp');
 	expect(transport2.sctpParameters).toBeUndefined();
 	expect(transport2.sctpState).toBeUndefined();

diff --git a/rust/src/router/plain_transport.rs b/rust/src/router/plain_transport.rs
@@ -44,6 +44,8 @@ use std::sync::{Arc, Weak};
 pub struct PlainTransportOptions {
     /// Listening info.
     pub listen_info: ListenInfo,
+    /// Optional listening info for RTCP.
+    pub rtcp_listen_info: Option<ListenInfo>,
     /// Use RTCP-mux (RTP and RTCP in the same port).
     /// Default true.
     pub rtcp_mux: bool,
@@ -79,6 +81,7 @@ impl PlainTransportOptions {
     pub fn new(listen_info: ListenInfo) -> Self {
         Self {
             listen_info,
+            rtcp_listen_info: None,
             rtcp_mux: true,
             comedia: false,
             enable_sctp: false,

diff --git a/rust/tests/integration/plain_transport.rs b/rust/tests/integration/plain_transport.rs
@@ -197,6 +197,7 @@ fn create_succeeds() {
         }
 
         {
+            let rtcp_port = pick_unused_port().unwrap();
             let transport2 = router
                 .create_plain_transport({
                     let mut plain_transport_options = PlainTransportOptions::new(ListenInfo {
@@ -209,6 +210,15 @@ fn create_succeeds() {
                     });
                     plain_transport_options.rtcp_mux = false;
 
+                    plain_transport_options.rtcp_listen_info = Some(ListenInfo {
+                        protocol: Protocol::Udp,
+                        ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+                        announced_ip: None,
+                        port: Some(rtcp_port),
+                        send_buffer_size: None,
+                        recv_buffer_size: None,
+                    });
+
                     plain_transport_options
                 })
                 .await
@@ -229,10 +239,14 @@ fn create_succeeds() {
             }
             assert!(transport2.rtcp_tuple().is_some());
             if let TransportTuple::LocalOnly {
-                local_ip, protocol, ..
+                local_ip,
+                local_port,
+                protocol,
+                ..
             } = transport2.rtcp_tuple().unwrap()
             {
                 assert_eq!(local_ip, "127.0.0.1".parse::<IpAddr>().unwrap());
+                assert_eq!(local_port, rtcp_port);
                 assert_eq!(protocol, Protocol::Udp);
             }
             assert_eq!(transport2.srtp_parameters(), None);

diff --git a/worker/fbs/plainTransport.fbs b/worker/fbs/plainTransport.fbs
@@ -6,6 +6,7 @@ namespace FBS.PlainTransport;
 table PlainTransportOptions {
   base:FBS.Transport.Options;
   listen_info:FBS.Transport.ListenInfo (required);
+  rtcp_listen_info:FBS.Transport.ListenInfo;
   rtcp_mux:bool;
   comedia:bool;
   enable_srtp:bool;

diff --git a/worker/include/RTC/PlainTransport.hpp b/worker/include/RTC/PlainTransport.hpp
@@ -77,6 +77,7 @@ namespace RTC
 		RTC::SrtpSession* srtpSendSession{ nullptr };
 		// Others.
 		ListenInfo listenInfo;
+		ListenInfo rtcpListenInfo;
 		bool rtcpMux{ true };
 		bool comedia{ false };
 		struct sockaddr_storage remoteAddrStorage;

diff --git a/worker/src/RTC/PipeTransport.cpp b/worker/src/RTC/PipeTransport.cpp
@@ -88,13 +88,13 @@ namespace RTC
 			if (this->listenInfo.sendBufferSize != 0)
 			{
 				// NOTE: This may throw.
-				udpSocket->SetSendBufferSize(this->listenInfo.sendBufferSize);
+				this->udpSocket->SetSendBufferSize(this->listenInfo.sendBufferSize);
 			}
 
 			if (this->listenInfo.recvBufferSize != 0)
 			{
 				// NOTE: This may throw.
-				udpSocket->SetRecvBufferSize(this->listenInfo.recvBufferSize);
+				this->udpSocket->SetRecvBufferSize(this->listenInfo.recvBufferSize);
 			}
 
 			MS_DEBUG_TAG(

diff --git a/worker/src/RTC/PlainTransport.cpp b/worker/src/RTC/PlainTransport.cpp
@@ -83,6 +83,44 @@ namespace RTC
 		this->rtcpMux = options->rtcpMux();
 		this->comedia = options->comedia();
 
+		if (!this->rtcpMux)
+		{
+			if (flatbuffers::IsFieldPresent(options, FBS::PlainTransport::PlainTransportOptions::VT_RTCPLISTENINFO))
+			{
+				if (options->rtcpListenInfo()->protocol() != FBS::Transport::Protocol::UDP)
+				{
+					MS_THROW_TYPE_ERROR("unsupported RTCP listen protocol");
+				}
+
+				this->rtcpListenInfo.ip.assign(options->rtcpListenInfo()->ip()->str());
+
+				// This may throw.
+				Utils::IP::NormalizeIp(this->rtcpListenInfo.ip);
+
+				if (flatbuffers::IsFieldPresent(options->rtcpListenInfo(), FBS::Transport::ListenInfo::VT_ANNOUNCEDIP))
+				{
+					this->rtcpListenInfo.announcedIp.assign(options->rtcpListenInfo()->announcedIp()->str());
+				}
+
+				this->rtcpListenInfo.port = options->rtcpListenInfo()->port();
+
+				if (flatbuffers::IsFieldPresent(options->rtcpListenInfo(), FBS::Transport::ListenInfo::VT_SENDBUFFERSIZE))
+				{
+					this->rtcpListenInfo.sendBufferSize = options->rtcpListenInfo()->sendBufferSize();
+				}
+
+				if (flatbuffers::IsFieldPresent(options->rtcpListenInfo(), FBS::Transport::ListenInfo::VT_RECVBUFFERSIZE))
+				{
+					this->rtcpListenInfo.recvBufferSize = options->rtcpListenInfo()->recvBufferSize();
+				}
+			}
+			// If rtcpListenInfo is not given, just clone listenInfo.
+			else
+			{
+				this->rtcpListenInfo = this->listenInfo;
+			}
+		}
+
 		if (options->enableSrtp())
 		{
 			if (!flatbuffers::IsFieldPresent(
@@ -151,30 +189,37 @@ namespace RTC
 			if (this->listenInfo.sendBufferSize != 0)
 			{
 				// NOTE: This may throw.
-				udpSocket->SetSendBufferSize(this->listenInfo.sendBufferSize);
+				this->udpSocket->SetSendBufferSize(this->listenInfo.sendBufferSize);
 			}
 
 			if (this->listenInfo.recvBufferSize != 0)
 			{
 				// NOTE: This may throw.
-				udpSocket->SetRecvBufferSize(this->listenInfo.recvBufferSize);
+				this->udpSocket->SetRecvBufferSize(this->listenInfo.recvBufferSize);
 			}
 
 			if (!this->rtcpMux)
 			{
 				// This may throw.
-				this->rtcpUdpSocket = new RTC::UdpSocket(this, this->listenInfo.ip);
+				if (this->rtcpListenInfo.port != 0)
+				{
+					this->rtcpUdpSocket = new RTC::UdpSocket(this, this->rtcpListenInfo.ip, this->rtcpListenInfo.port);
+				}
+				else
+				{
+					this->rtcpUdpSocket = new RTC::UdpSocket(this, this->rtcpListenInfo.ip);
+				}
 
-				if (this->listenInfo.sendBufferSize != 0)
+				if (this->rtcpListenInfo.sendBufferSize != 0)
 				{
 					// NOTE: This may throw.
-					rtcpUdpSocket->SetSendBufferSize(this->listenInfo.sendBufferSize);
+					this->rtcpUdpSocket->SetSendBufferSize(this->rtcpListenInfo.sendBufferSize);
 				}
 
-				if (this->listenInfo.recvBufferSize != 0)
+				if (this->rtcpListenInfo.recvBufferSize != 0)
 				{
 					// NOTE: This may throw.
-					rtcpUdpSocket->SetRecvBufferSize(this->listenInfo.recvBufferSize);
+					this->rtcpUdpSocket->SetRecvBufferSize(this->rtcpListenInfo.recvBufferSize);
 				}
 			}
 
@@ -263,13 +308,13 @@ namespace RTC
 			{
 				std::string localIp;
 
-				if (this->listenInfo.announcedIp.empty())
+				if (this->rtcpListenInfo.announcedIp.empty())
 				{
 					localIp = this->rtcpUdpSocket->GetLocalIp();
 				}
 				else
 				{
-					localIp = this->listenInfo.announcedIp;
+					localIp = this->rtcpListenInfo.announcedIp;
 				}
 
 				rtcpTuple = FBS::Transport::CreateTupleDirect(
@@ -645,9 +690,9 @@ namespace RTC
 							  this->rtcpUdpSocket,
 							  reinterpret_cast<struct sockaddr*>(&this->rtcpRemoteAddrStorage));
 
-							if (!this->listenInfo.announcedIp.empty())
+							if (!this->rtcpListenInfo.announcedIp.empty())
 							{
-								this->rtcpTuple->SetLocalAnnouncedIp(this->listenInfo.announcedIp);
+								this->rtcpTuple->SetLocalAnnouncedIp(this->rtcpListenInfo.announcedIp);
 							}
 						}
 					}
@@ -1076,9 +1121,9 @@ namespace RTC
 
 			this->rtcpTuple = new RTC::TransportTuple(tuple);
 
-			if (!this->listenInfo.announcedIp.empty())
+			if (!this->rtcpListenInfo.announcedIp.empty())
 			{
-				this->rtcpTuple->SetLocalAnnouncedIp(this->listenInfo.announcedIp);
+				this->rtcpTuple->SetLocalAnnouncedIp(this->rtcpListenInfo.announcedIp);
 			}
 
 			// Notify the Node PlainTransport.