muxers: Add test harness for StreamMuxer implementations

Cargo.toml

@@ -142,6 +142,7 @@ members = [
     "misc/quickcheck-ext",
     "muxers/mplex",
     "muxers/yamux",
+    "muxers/test-harness",
     "protocols/dcutr",
     "protocols/autonat",
     "protocols/floodsub",

core/CHANGELOG.md

@@ -5,8 +5,11 @@
 - Remove default features. If you previously depended on `secp256k1` or `ecdsa` you need to enable these explicitly 
   now. See [PR 2918].
 
+- Deprecate `StreamMuxerExt::next_{inbound,outbound}`. See [PR XXXX].
+
 [PR 2915]: https://github.com/libp2p/rust-libp2p/pull/2915
 [PR 2918]: https://github.com/libp2p/rust-libp2p/pull/2918
+[PR XXXX]: https://github.com/libp2p/rust-libp2p/pull/XXXX
 
 # 0.36.0
 

core/src/muxing.rs

@@ -160,11 +160,19 @@ pub trait StreamMuxerExt: StreamMuxer + Sized {
     }
 
     /// Returns a future that resolves to the next inbound `Substream` opened by the remote.
+    #[deprecated(
+        since = "0.36.1",
+        note = "This future violates the `StreamMuxer` contract because it doesn't call `StreamMuxer::poll`."
+    )]
     fn next_inbound(&mut self) -> NextInbound<'_, Self> {
         NextInbound(self)
     }
 
     /// Returns a future that opens a new outbound `Substream` with the remote.
+    #[deprecated(
+        since = "0.36.1",
+        note = "This future violates the `StreamMuxer` contract because it doesn't call `StreamMuxer::poll`."
+    )]
     fn next_outbound(&mut self) -> NextOutbound<'_, Self> {
         NextOutbound(self)
     }

muxers/mplex/Cargo.toml

@@ -23,11 +23,12 @@ smallvec = "1.6.1"
 unsigned-varint = { version = "0.7", features = ["asynchronous_codec"] }
 
 [dev-dependencies]
-async-std = "1.7.0"
+async-std = { version = "1.7.0", features = ["attributes"] }
 criterion = "0.4"
 env_logger = "0.9"
 futures = "0.3"
 libp2p = { path = "../..", features = ["full"] }
+libp2p-muxer-test-harness = { path = "../test-harness" }
 quickcheck = { package = "quickcheck-ext", path = "../../misc/quickcheck-ext" }
 
 [[bench]]

muxers/mplex/benches/split_send_size.rs

@@ -114,7 +114,10 @@ fn run(
                 }
                 transport::TransportEvent::Incoming { upgrade, .. } => {
                     let (_peer, mut conn) = upgrade.await.unwrap();
-                    let mut s = conn.next_inbound().await.expect("unexpected error");
+                    // Just calling `poll_inbound` without `poll` is fine here because mplex makes progress through all `poll_` functions. It is hacky though.
+                    let mut s = poll_fn(|cx| conn.poll_inbound_unpin(cx))
+                        .await
+                        .expect("unexpected error");
 
                     let mut buf = vec![0u8; payload_len];
                     let mut off = 0;
@@ -139,7 +142,8 @@ fn run(
     let sender = async move {
         let addr = addr_receiver.await.unwrap();
         let (_peer, mut conn) = sender_trans.dial(addr).unwrap().await.unwrap();
-        let mut stream = conn.next_outbound().await.unwrap();
+        // Just calling `poll_outbound` without `poll` is fine here because mplex makes progress through all `poll_` functions. It is hacky though.
+        let mut stream = poll_fn(|cx| conn.poll_outbound_unpin(cx)).await.unwrap();
         let mut off = 0;
         loop {
             let n = poll_fn(|cx| Pin::new(&mut stream).poll_write(cx, &payload[off..]))

muxers/mplex/tests/compliance.rs

@@ -0,0 +1,37 @@
+use libp2p_mplex::MplexConfig;
+
+#[async_std::test]
+async fn close_implies_flush() {
+    let (alice, bob) =
+        libp2p_muxer_test_harness::connected_muxers_on_memory_transport::<MplexConfig, _, _>()
+            .await;
+
+    libp2p_muxer_test_harness::close_implies_flush(alice, bob).await;
+}
+
+#[async_std::test]
+async fn dialer_can_receive() {
+    let (alice, bob) =
+        libp2p_muxer_test_harness::connected_muxers_on_memory_transport::<MplexConfig, _, _>()
+            .await;
+
+    libp2p_muxer_test_harness::dialer_can_receive(alice, bob).await;
+}
+
+#[async_std::test]
+async fn read_after_close() {
+    let (alice, bob) =
+        libp2p_muxer_test_harness::connected_muxers_on_memory_transport::<MplexConfig, _, _>()
+            .await;
+
+    libp2p_muxer_test_harness::read_after_close(alice, bob).await;
+}
+
+#[async_std::test]
+async fn write_to_remote_closed() {
+    let (alice, bob) =
+        libp2p_muxer_test_harness::connected_muxers_on_memory_transport::<MplexConfig, _, _>()
+            .await;
+
+    libp2p_muxer_test_harness::write_to_remote_closed(alice, bob).await;
+}