Add compliance test to mplex that dropping a stream resets it

muxers/mplex/tests/compliance.rs

@@ -26,3 +26,14 @@ async fn read_after_close() {
 
     libp2p_muxer_test_harness::read_after_close(alice, bob).await;
 }
+
+#[async_std::test]
+async fn drop_resets_stream() {
+    env_logger::init();
+
+    let (alice, bob) =
+        libp2p_muxer_test_harness::connected_muxers_on_memory_transport::<MplexConfig, _, _>()
+            .await;
+
+    libp2p_muxer_test_harness::drop_resets_stream(alice, bob).await;
+}

muxers/test-harness/src/lib.rs

@@ -138,6 +138,30 @@ where
     .await;
 }
 
+/// Verifies that dropping a substream causes a reset.
+pub async fn drop_resets_stream<A, B, S, E>(alice: A, bob: B)
+where
+    A: StreamMuxer<Substream = S, Error = E> + Unpin,
+    B: StreamMuxer<Substream = S, Error = E> + Unpin,
+    S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    E: fmt::Debug,
+{
+    run_commutative(
+        alice,
+        bob,
+        |stream| async move {
+            drop(stream);
+        },
+        |mut stream| async move {
+            let mut buf = [0u8; 4];
+            let error = stream.read_exact(&mut buf).await.unwrap_err();
+
+            assert_eq!(error.kind(), io::ErrorKind::ConnectionReset);
+        },
+    )
+    .await;
+}
+
 /// Runs the given protocol between the two parties, ensuring commutativity, i.e. either party can be the dialer and listener.
 async fn run_commutative<A, B, S, E, F1, F2>(
     mut alice: A,