chore(multistream-select): replace async-std with tokio in tests

Fixes #4449

Pull-Request: #6057.

Author: PoulavBhowmick03

Cargo.lock

@@ -19,11 +19,12 @@ smallvec = "1.13.2"
 unsigned-varint = { workspace = true }
 
 [dev-dependencies]
-async-std = { version = "1.6.2", features = ["attributes"] }
 futures_ringbuf = "0.4.0"
 quickcheck = { workspace = true }
 rw-stream-sink = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+tokio = { workspace = true, features = ["full"] }
+tokio-util = { version = "0.7", features = ["compat"] }
 
 # Passing arguments to the docsrs builder in order to properly document cfg's.
 # More information: https://docs.rs/about/builds#cross-compiling

misc/multistream-select/Cargo.toml

@@ -208,13 +208,16 @@ where
 
 #[cfg(test)]
 mod tests {
+
     use std::time::Duration;
 
-    use async_std::{
-        future::timeout,
+    use quickcheck::{Arbitrary, Gen, GenRange};
+    use tokio::{
         net::{TcpListener, TcpStream},
+        runtime::Runtime,
+        time::timeout,
     };
-    use quickcheck::{Arbitrary, Gen, GenRange};
+    use tokio_util::compat::TokioAsyncReadCompatExt;
     use tracing::metadata::LevelFilter;
     use tracing_subscriber::EnvFilter;
 
@@ -226,7 +229,7 @@ mod tests {
         async fn run(version: Version) {
             let (client_connection, server_connection) = futures_ringbuf::Endpoint::pair(100, 100);
 
-            let server = async_std::task::spawn(async move {
+            let server = tokio::task::spawn(async move {
                 let protos = vec!["/proto1", "/proto2"];
                 let (proto, mut io) = listener_select_proto(server_connection, protos)
                     .await
@@ -242,7 +245,7 @@ mod tests {
                 io.flush().await.unwrap();
             });
 
-            let client = async_std::task::spawn(async move {
+            let client = tokio::task::spawn(async move {
                 let protos = vec!["/proto3", "/proto2"];
                 let (proto, mut io) = dialer_select_proto(client_connection, protos, version)
                     .await
@@ -258,12 +261,13 @@ mod tests {
                 assert_eq!(out, b"pong");
             });
 
-            server.await;
-            client.await;
+            server.await.unwrap();
+            client.await.unwrap();
         }
 
-        async_std::task::block_on(run(Version::V1));
-        async_std::task::block_on(run(Version::V1Lazy));
+        let rt = Runtime::new().unwrap();
+        rt.block_on(run(Version::V1));
+        rt.block_on(run(Version::V1Lazy));
     }
 
     /// Tests the expected behaviour of failed negotiations.
@@ -283,12 +287,13 @@ mod tests {
                 )
                 .try_init();
 
-            async_std::task::block_on(async move {
+            let rt = Runtime::new().unwrap();
+            rt.block_on(async move {
                 let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
                 let addr = listener.local_addr().unwrap();
 
-                let server = async_std::task::spawn(async move {
-                    let server_connection = listener.accept().await.unwrap().0;
+                let server = tokio::task::spawn(async move {
+                    let server_connection = listener.accept().await.unwrap().0.compat();
 
                     let io = match timeout(
                         Duration::from_secs(2),
@@ -309,8 +314,8 @@ mod tests {
                     }
                 });
 
-                let client = async_std::task::spawn(async move {
-                    let client_connection = TcpStream::connect(addr).await.unwrap();
+                let client = tokio::task::spawn(async move {
+                    let client_connection = TcpStream::connect(addr).await.unwrap().compat();
 
                     let mut io = match timeout(
                         Duration::from_secs(2),
@@ -336,8 +341,8 @@ mod tests {
                     }
                 });
 
-                server.await;
-                client.await;
+                server.await.unwrap();
+                client.await.unwrap();
 
                 tracing::info!("---------------------------------------")
             });
@@ -348,12 +353,12 @@ mod tests {
             .quickcheck(prop as fn(_, _, _, _));
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn v1_lazy_do_not_wait_for_negotiation_on_poll_close() {
         let (client_connection, _server_connection) =
             futures_ringbuf::Endpoint::pair(1024 * 1024, 1);
 
-        let client = async_std::task::spawn(async move {
+        let client = tokio::task::spawn(async move {
             // Single protocol to allow for lazy (or optimistic) protocol negotiation.
             let protos = vec!["/proto1"];
             let (proto, mut io) = dialer_select_proto(client_connection, protos, Version::V1Lazy)
@@ -366,9 +371,12 @@ mod tests {
             io.close().await.unwrap();
         });
 
-        async_std::future::timeout(Duration::from_secs(10), client)
-            .await
-            .unwrap();
+        match tokio::time::timeout(Duration::from_secs(10), client).await {
+            Ok(join_result) => join_result.expect("Client task should complete successfully"),
+            Err(_elapsed) => {
+                panic!("Expected the client task to complete before timeout");
+            }
+        }
     }
 
     #[derive(Clone, Debug)]

misc/multistream-select/src/dialer_select.rs

@@ -388,6 +388,7 @@ mod tests {
 
     use futures::{io::Cursor, prelude::*};
     use quickcheck::*;
+    use tokio::runtime::Runtime;
 
     use crate::length_delimited::LengthDelimited;
 
@@ -491,9 +492,10 @@ mod tests {
         fn prop(frames: Vec<Vec<u8>>) -> TestResult {
             let (client_connection, server_connection) = futures_ringbuf::Endpoint::pair(100, 100);
 
-            async_std::task::block_on(async move {
+            let rt = Runtime::new().unwrap();
+            rt.block_on(async move {
                 let expected_frames = frames.clone();
-                let server = async_std::task::spawn(async move {
+                let server = tokio::task::spawn(async move {
                     let mut connec =
                         rw_stream_sink::RwStreamSink::new(LengthDelimited::new(server_connection));
 
@@ -510,15 +512,15 @@ mod tests {
                     }
                 });
 
-                let client = async_std::task::spawn(async move {
+                let client = tokio::task::spawn(async move {
                     let mut connec = LengthDelimited::new(client_connection);
                     for frame in frames {
                         connec.send(From::from(frame)).await.unwrap();
                     }
                 });
 
-                server.await;
-                client.await;
+                server.await.unwrap();
+                client.await.unwrap();
             });
 
             TestResult::passed()

misc/multistream-select/src/length_delimited.rs

@@ -69,17 +69,19 @@
 //! For a dialer:
 //!
 //! ```no_run
-//! use async_std::net::TcpStream;
 //! use futures::prelude::*;
 //! use multistream_select::{dialer_select_proto, Version};
+//! use tokio::{net::TcpStream, runtime::Runtime};
+//! use tokio_util::compat::TokioAsyncReadCompatExt;
 //!
-//! async_std::task::block_on(async move {
+//! let rt = Runtime::new().unwrap();
+//! rt.block_on(async move {
 //!     let socket = TcpStream::connect("127.0.0.1:10333").await.unwrap();
+//!     let compat_socket = socket.compat();
 //!
 //!     let protos = vec!["/echo/1.0.0", "/echo/2.5.0"];
-//!     let (protocol, _io) = dialer_select_proto(socket, protos, Version::V1)
-//!         .await
-//!         .unwrap();
+//!     let result = dialer_select_proto(compat_socket, protos, Version::V1).await;
+//!     let (protocol, _io) = result.unwrap();
 //!
 //!     println!("Negotiated protocol: {:?}", protocol);
 //!     // You can now use `_io` to communicate with the remote.