feat: add cancellation token support to connection methods across transport layers

Author: shenjinti

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsipstack"
-version = "0.2.26"
+version = "0.2.27"
 edition = "2021"
 description = "SIP Stack Rust library for building SIP applications"
 license = "MIT"

examples/client/main.rs

@@ -174,6 +174,7 @@ async fn main() -> rsipstack::Result<()> {
     let mut connection = UdpConnection::create_connection(
         format!("{}:{}", addr, args.port).parse()?,
         external.clone(),
+        Some(token.child_token()),
     )
     .await?;
 

examples/client/play_file.rs

@@ -21,7 +21,8 @@ pub async fn build_rtp_conn(
     for p in 0..100 {
         let port = opt.rtp_start_port + p * 2;
         if let Ok(c) =
-            UdpConnection::create_connection(format!("{:?}:{}", addr, port).parse()?, None).await
+            UdpConnection::create_connection(format!("{:?}:{}", addr, port).parse()?, None, None)
+                .await
         {
             conn = Some(c);
             break;

examples/proxy.rs

@@ -128,6 +128,7 @@ async fn main() -> Result<()> {
     let connection = UdpConnection::create_connection(
         format!("{}:{}", addr, args.port).parse()?,
         external.clone(),
+        None,
     )
     .await?;
     transport_layer.add_transport(connection.into());
@@ -630,12 +631,13 @@ async fn handle_websocket(client_addr: ClientAddr, socket: WebSocket, _state: Ap
         r#type: Some(transport_type),
         addr: client_addr.0.into(),
     };
-
+    let ws_token = CancellationToken::new();
     // Create the ChannelConnection
     let connection = match ChannelConnection::create_connection(
         from_ws_rx,
         to_ws_tx,
         local_addr.clone(),
+        Some(ws_token),
     )
     .await
     {
@@ -784,7 +786,7 @@ mod tests {
         let token = CancellationToken::new();
         let transport_layer = TransportLayer::new(token.clone());
 
-        let udp_conn = UdpConnection::create_connection("127.0.0.1:0".parse().unwrap(), None)
+        let udp_conn = UdpConnection::create_connection("127.0.0.1:0".parse().unwrap(), None, None)
             .await
             .unwrap();
         transport_layer.add_transport(udp_conn.into());
@@ -906,7 +908,7 @@ mod tests {
         let transport_layer = TransportLayer::new(token.clone());
 
         // Test that we can add UDP transport
-        let udp_conn = UdpConnection::create_connection("127.0.0.1:0".parse().unwrap(), None)
+        let udp_conn = UdpConnection::create_connection("127.0.0.1:0".parse().unwrap(), None, None)
             .await
             .unwrap();
         transport_layer.add_transport(udp_conn.into());
@@ -945,6 +947,7 @@ mod tests {
             to_transport_rx,
             from_transport_tx.clone(),
             local_addr.clone(),
+            None,
         )
         .await
         .expect("Should create channel connection");

src/bin/bench_ua.rs

@@ -358,7 +358,8 @@ async fn main() -> Result<()> {
 
     // Setup UDP connection
     let addr = format!("0.0.0.0:{}", args.port);
-    let connection = UdpConnection::create_connection(addr.parse()?, None).await?;
+    let connection =
+        UdpConnection::create_connection(addr.parse()?, None, Some(token.child_token())).await?;
     transport_layer.add_transport(connection.into());
 
     let endpoint = EndpointBuilder::new()

src/dialog/tests/test_dialog_layer.rs

@@ -50,8 +50,7 @@ fn create_invite_request(from_tag: &str, to_tag: &str, call_id: &str, branch: &s
 
 /// Test helper to create mock connection
 async fn create_mock_connection() -> crate::Result<crate::transport::SipConnection> {
-    // 使用随机端口创建UDP连接
-    let udp_conn = UdpConnection::create_connection("127.0.0.1:0".parse()?, None).await?;
+    let udp_conn = UdpConnection::create_connection("127.0.0.1:0".parse()?, None, None).await?;
     Ok(udp_conn.into())
 }
 

src/lib.rs

@@ -162,7 +162,7 @@
 //! # async fn example() -> rsipstack::Result<()> {
 //! # let cancel_token = CancellationToken::new();
 //! let transport_layer = TransportLayer::new(cancel_token.child_token());
-//! let udp_conn = UdpConnection::create_connection("0.0.0.0:5060".parse()?, None).await?;
+//! let udp_conn = UdpConnection::create_connection("0.0.0.0:5060".parse()?, None, Some(cancel_token.child_token())).await?;
 //! transport_layer.add_transport(udp_conn.into());
 //!
 //! let endpoint = EndpointBuilder::new()

src/transaction/tests/mod.rs

@@ -15,7 +15,7 @@ pub(super) async fn create_test_endpoint(addr: Option<&str>) -> Result<Endpoint>
     let tl = TransportLayer::new(token.child_token());
 
     if let Some(addr) = addr {
-        let peer = UdpConnection::create_connection(addr.parse()?, None).await?;
+        let peer = UdpConnection::create_connection(addr.parse()?, None, None).await?;
         tl.add_transport(peer.into());
     }
 

src/transaction/tests/test_client.rs

@@ -17,7 +17,7 @@ async fn test_client_transaction() -> Result<()> {
         .to_owned();
     info!("server addr: {}", server_addr);
 
-    let peer_server = UdpConnection::create_connection("127.0.0.1:0".parse()?, None).await?;
+    let peer_server = UdpConnection::create_connection("127.0.0.1:0".parse()?, None, None).await?;
     let peer_server_loop = async {
         let (sender, mut recevier) = unbounded_channel();
         select! {

src/transaction/tests/test_endpoint.rs

@@ -57,6 +57,7 @@ async fn test_endpoint_recvrequests() {
         let test_conn = crate::transport::udp::UdpConnection::create_connection(
             "127.0.0.1:0".parse().unwrap(),
             None,
+            None,
         )
         .await
         .expect("create_connection");