Add tests for MultiPacket channels

Author: leynos

src/response.rs

@@ -50,6 +50,42 @@ pub enum Response<F, E = ()> {
     /// A potentially unbounded stream of frames.
     Stream(FrameStream<F, E>),
     /// Frames delivered through a channel.
+    ///
+    /// # Usage and lifecycle
+    ///
+    /// `MultiPacket` wraps a [`tokio::sync::mpsc::Receiver`] that yields frames
+    /// (`F`) sent from another task. The receiver should be polled until it
+    /// returns `None`, signalling the channel has closed and no more frames will
+    /// be sent. Frames are yielded in send order.
+    /// Back-pressure follows the channel's capacity: senders await when it is
+    /// full. The stream ends once all senders are dropped and `recv` returns
+    /// `None`.
+    ///
+    /// # Resource management
+    ///
+    /// To avoid resource leaks or deadlocks:
+    /// - Drop the sender once all frames are sent.
+    /// - Poll the receiver to completion, consuming all frames.
+    /// - If the sender is dropped early, the receiver yields `None` and no further frames will
+    ///   arrive.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use tokio::sync::mpsc;
+    /// use wireframe::Response;
+    ///
+    /// # tokio_test::block_on(async {
+    /// let (tx, rx) = mpsc::channel(1);
+    /// tx.send(1u8).await.expect("send");
+    /// drop(tx);
+    /// if let Response::MultiPacket(mut rx) = Response::MultiPacket(rx) {
+    ///     while let Some(f) = rx.recv().await {
+    ///         assert_eq!(f, 1);
+    ///     }
+    /// }
+    /// # });
+    /// ```
     MultiPacket(mpsc::Receiver<F>),
     /// A response with no frames.
     Empty,

tests/features/multi_packet.feature

@@ -2,3 +2,7 @@ Feature: Multi-packet responses
   Scenario: messages from a multi-packet response are delivered sequentially
     When a multi-packet response emits messages
     Then all messages are received in order
+
+  Scenario: no messages are emitted from a multi-packet response
+    When a multi-packet response emits no messages
+    Then no messages are received

tests/multi_packet.rs

@@ -23,3 +23,57 @@ async fn multi_packet_yields_messages() {
     }
     assert_eq!(received, vec![TestMsg(1), TestMsg(2)]);
 }
+
+/// Yields no messages when the channel is immediately closed.
+#[tokio::test]
+async fn multi_packet_empty_channel() {
+    let (_tx, rx) = mpsc::channel(4);
+    drop(_tx);
+    let resp: Response<TestMsg, ()> = Response::MultiPacket(rx);
+    let mut received = Vec::new();
+    if let Response::MultiPacket(mut rx) = resp {
+        while let Some(msg) = rx.recv().await {
+            received.push(msg);
+        }
+    }
+    assert!(received.is_empty());
+}
+
+/// Stops yielding when the sender is dropped before all messages are sent.
+#[tokio::test]
+async fn multi_packet_sender_dropped_before_all_messages() {
+    let (tx, rx) = mpsc::channel(4);
+    tx.send(TestMsg(1)).await.expect("send");
+    drop(tx);
+    let resp: Response<TestMsg, ()> = Response::MultiPacket(rx);
+    let mut received = Vec::new();
+    if let Response::MultiPacket(mut rx) = resp {
+        while let Some(msg) = rx.recv().await {
+            received.push(msg);
+        }
+    }
+    assert_eq!(received, vec![TestMsg(1)]);
+}
+
+/// Handles more messages than the channel capacity allows.
+#[tokio::test]
+async fn multi_packet_handles_channel_capacity() {
+    let (tx, rx) = mpsc::channel(2);
+    let send_task = tokio::spawn(async move {
+        for i in 0..4u8 {
+            tx.send(TestMsg(i)).await.expect("send");
+        }
+    });
+    let resp: Response<TestMsg, ()> = Response::MultiPacket(rx);
+    let mut received = Vec::new();
+    if let Response::MultiPacket(mut rx) = resp {
+        while let Some(msg) = rx.recv().await {
+            received.push(msg);
+        }
+    }
+    send_task.await.expect("sender join");
+    assert_eq!(
+        received,
+        vec![TestMsg(0), TestMsg(1), TestMsg(2), TestMsg(3)]
+    );
+}

tests/steps/multi_packet_steps.rs

@@ -8,3 +8,9 @@ async fn when_multi(world: &mut MultiPacketWorld) { world.process().await; }
 
 #[then("all messages are received in order")]
 fn then_multi(world: &mut MultiPacketWorld) { world.verify(); }
+
+#[when("a multi-packet response emits no messages")]
+async fn when_multi_empty(world: &mut MultiPacketWorld) { world.process_empty().await; }
+
+#[then("no messages are received")]
+fn then_multi_empty(world: &mut MultiPacketWorld) { world.verify_empty(); }

tests/world.rs

@@ -149,11 +149,6 @@ impl CorrelationWorld {
     ///
     /// # Panics
     /// Panics if the actor fails to run successfully.
-    /// Send messages through a multi-packet response and record them.
-    ///
-    /// # Panics
-    ///
-    /// Panics if sending to the channel fails.
     pub async fn process(&mut self) {
         let cid = self.cid;
         let stream: FrameStream<Envelope> = Box::pin(try_stream! {
@@ -166,16 +161,10 @@ impl CorrelationWorld {
         actor.run(&mut self.frames).await.expect("actor run failed");
     }
 
-    /// Verify that all received frames carry the expected correlation id.
-    ///
-    /// # Panics
-    /// Panics if any frame has a `correlation_id` that does not match the
-    /// expected value.
-    /// Verify messages were received in order.
+    /// Verify that all received frames carry the expected correlation ID.
     ///
     /// # Panics
-    ///
-    /// Panics if the messages are not in the expected order.
+    /// Panics if any frame has a `correlation_id` that does not match `self.cid`.
     pub fn verify(&self) {
         assert!(
             self.frames
@@ -197,11 +186,6 @@ impl StreamEndWorld {
     ///
     /// # Panics
     /// Panics if the actor fails to run successfully.
-    /// Send messages through a multi-packet response and record them.
-    ///
-    /// # Panics
-    ///
-    /// Panics if sending to the channel fails.
     pub async fn process(&mut self) {
         let stream: FrameStream<u8> = Box::pin(try_stream! {
             yield 1u8;
@@ -233,22 +217,44 @@ impl MultiPacketWorld {
     /// Send messages through a multi-packet response and record them.
     ///
     /// # Panics
-    ///
     /// Panics if sending to the channel fails.
     pub async fn process(&mut self) {
-        let (tx, rx) = tokio::sync::mpsc::channel(4);
+        let (tx, ch_rx) = tokio::sync::mpsc::channel(4);
         tx.send(1u8).await.expect("send");
         tx.send(2u8).await.expect("send");
         tx.send(3u8).await.expect("send");
         drop(tx);
-        let resp: wireframe::Response<u8, ()> = wireframe::Response::MultiPacket(rx);
-        if let wireframe::Response::MultiPacket(mut rx) = resp {
-            while let Some(msg) = rx.recv().await {
+        let resp: wireframe::Response<u8, ()> = wireframe::Response::MultiPacket(ch_rx);
+        if let wireframe::Response::MultiPacket(mut mp_rx) = resp {
+            while let Some(msg) = mp_rx.recv().await {
+                self.messages.push(msg);
+            }
+        }
+    }
+
+    /// Send no messages through the channel and record them.
+    ///
+    /// # Panics
+    /// Panics if sending to the channel fails.
+    pub async fn process_empty(&mut self) {
+        let (_tx, ch_rx) = tokio::sync::mpsc::channel(4);
+        drop(_tx);
+        let resp: wireframe::Response<u8, ()> = wireframe::Response::MultiPacket(ch_rx);
+        if let wireframe::Response::MultiPacket(mut mp_rx) = resp {
+            while let Some(msg) = mp_rx.recv().await {
                 self.messages.push(msg);
             }
         }
     }
 
+    /// Verify that no messages were received.
+    ///
+    /// # Panics
+    /// Panics if any messages are present.
+    pub fn verify_empty(&self) {
+        assert!(self.messages.is_empty());
+    }
+
     /// Verify messages were received in order.
     ///
     /// # Panics