feat: rtp

Cargo.toml

@@ -40,6 +40,7 @@ once_cell = "1"
 pin-project-lite = "0.2"
 rand = "0.8"
 regex = "1"
+rtp = { git = "https://github.com/Banyc/rtp.git", tag = "v0.0.4" }
 scopeguard = "1"
 serde = "1"
 serde_json = "1"
@@ -59,7 +60,7 @@ tokio_throughput = { git = "https://github.com/Banyc/tokio_throughput.git", rev
 toml = "0.8"
 tracing = "0.1"
 tracing-subscriber = "0.3"
-udp_listener = { git = "https://github.com/Banyc/udp_listener.git", tag = "v0.0.6" }
+udp_listener = { git = "https://github.com/Banyc/udp_listener.git", tag = "v0.0.7" }
 
 [profile.profiling]
 inherits = "release"

protocol/Cargo.toml

@@ -12,6 +12,7 @@ metrics = { workspace = true }
 mptcp = { workspace = true }
 once_cell = { workspace = true }
 rand = { workspace = true }
+rtp = { workspace = true }
 serde = { workspace = true }
 swap = { workspace = true }
 thiserror = { workspace = true }

protocol/src/stream/addr.rs

@@ -16,6 +16,7 @@ pub enum ConcreteStreamType {
     Tcp,
     Kcp,
     Mptcp,
+    Rtp,
 }
 impl StreamType for ConcreteStreamType {}
 impl fmt::Display for ConcreteStreamType {
@@ -24,6 +25,7 @@ impl fmt::Display for ConcreteStreamType {
             ConcreteStreamType::Tcp => write!(f, "tcp"),
             ConcreteStreamType::Kcp => write!(f, "kcp"),
             ConcreteStreamType::Mptcp => write!(f, "mptcp"),
+            ConcreteStreamType::Rtp => write!(f, "rtp"),
         }
     }
 }
@@ -35,6 +37,7 @@ impl FromStr for ConcreteStreamType {
             "tcp" => Ok(Self::Tcp),
             "kcp" => Ok(Self::Kcp),
             "mptcp" => Ok(Self::Mptcp),
+            "rtp" => Ok(Self::Rtp),
             _ => Err(ParseInternetAddrError),
         }
     }

protocol/src/stream/connect.rs

@@ -5,14 +5,15 @@ use common::stream::connect::{StreamConnect, StreamConnectExt, StreamConnectorTa
 use super::{
     addr::ConcreteStreamType,
     connection::Connection,
-    streams::{kcp::KcpConnector, mptcp::MptcpConnector, tcp::TcpConnector},
+    streams::{kcp::KcpConnector, mptcp::MptcpConnector, rtp::RtpConnector, tcp::TcpConnector},
 };
 
 #[derive(Debug, Clone)]
 pub struct ConcreteStreamConnectorTable {
     tcp: TcpConnector,
     kcp: KcpConnector,
     mptcp: MptcpConnector,
+    rtp: RtpConnector,
 }
 
 impl ConcreteStreamConnectorTable {
@@ -21,6 +22,7 @@ impl ConcreteStreamConnectorTable {
             tcp: TcpConnector,
             kcp: KcpConnector,
             mptcp: MptcpConnector,
+            rtp: RtpConnector,
         }
     }
 }
@@ -38,6 +40,7 @@ impl StreamConnectorTable for ConcreteStreamConnectorTable {
             ConcreteStreamType::Tcp => self.tcp.connect(addr).await,
             ConcreteStreamType::Kcp => self.kcp.connect(addr).await,
             ConcreteStreamType::Mptcp => self.mptcp.connect(addr).await,
+            ConcreteStreamType::Rtp => self.rtp.connect(addr).await,
         }
     }
 
@@ -51,6 +54,7 @@ impl StreamConnectorTable for ConcreteStreamConnectorTable {
             ConcreteStreamType::Tcp => self.tcp.timed_connect(addr, timeout).await,
             ConcreteStreamType::Kcp => self.kcp.timed_connect(addr, timeout).await,
             ConcreteStreamType::Mptcp => self.mptcp.timed_connect(addr, timeout).await,
+            ConcreteStreamType::Rtp => self.rtp.timed_connect(addr, timeout).await,
         }
     }
 }

protocol/src/stream/connection.rs

@@ -6,7 +6,9 @@ use common::stream::{IoAddr, IoStream};
 
 use super::{
     addr::ConcreteStreamAddr,
-    streams::{kcp::AddressedKcpStream, mptcp::IoMptcpStream, tcp::IoTcpStream},
+    streams::{
+        kcp::AddressedKcpStream, mptcp::IoMptcpStream, rtp::AddressedRtpStream, tcp::IoTcpStream,
+    },
 };
 
 #[derive(Debug)]
@@ -15,6 +17,7 @@ pub enum Connection {
     Tcp(IoTcpStream),
     Kcp(AddressedKcpStream),
     Mptcp(IoMptcpStream),
+    Rtp(AddressedRtpStream),
 }
 
 impl IoStream for Connection {}
@@ -25,6 +28,7 @@ impl IoAddr for Connection {
             Connection::Tcp(x) => x.peer_addr(),
             Connection::Kcp(x) => x.peer_addr(),
             Connection::Mptcp(x) => IoAddr::peer_addr(x),
+            Connection::Rtp(x) => IoAddr::peer_addr(x),
         }
     }
 
@@ -34,6 +38,7 @@ impl IoAddr for Connection {
             Connection::Tcp(x) => x.local_addr(),
             Connection::Kcp(x) => x.local_addr(),
             Connection::Mptcp(x) => IoAddr::local_addr(x),
+            Connection::Rtp(x) => IoAddr::local_addr(x),
         }
     }
 }
@@ -49,6 +54,7 @@ impl AsyncWrite for Connection {
             Connection::Tcp(x) => Pin::new(x).poll_write(cx, buf),
             Connection::Kcp(x) => Pin::new(x).poll_write(cx, buf),
             Connection::Mptcp(x) => Pin::new(x).poll_write(cx, buf),
+            Connection::Rtp(x) => Pin::new(x).poll_write(cx, buf),
         }
     }
 
@@ -61,6 +67,7 @@ impl AsyncWrite for Connection {
             Connection::Tcp(x) => Pin::new(x).poll_flush(cx),
             Connection::Kcp(x) => Pin::new(x).poll_flush(cx),
             Connection::Mptcp(x) => Pin::new(x).poll_flush(cx),
+            Connection::Rtp(x) => Pin::new(x).poll_flush(cx),
         }
     }
 
@@ -73,6 +80,7 @@ impl AsyncWrite for Connection {
             Connection::Tcp(x) => Pin::new(x).poll_shutdown(cx),
             Connection::Kcp(x) => Pin::new(x).poll_shutdown(cx),
             Connection::Mptcp(x) => Pin::new(x).poll_shutdown(cx),
+            Connection::Rtp(x) => Pin::new(x).poll_shutdown(cx),
         }
     }
 }
@@ -88,6 +96,7 @@ impl AsyncRead for Connection {
             Connection::Tcp(x) => Pin::new(x).poll_read(cx, buf),
             Connection::Kcp(x) => Pin::new(x).poll_read(cx, buf),
             Connection::Mptcp(x) => Pin::new(x).poll_read(cx, buf),
+            Connection::Rtp(x) => Pin::new(x).poll_read(cx, buf),
         }
     }
 }

protocol/src/stream/streams/mod.rs

@@ -1,4 +1,5 @@
 pub mod kcp;
 // pub mod quic;
 pub mod mptcp;
+pub mod rtp;
 pub mod tcp;