From 9faaf8e38ef38d16dce7567742d68c2cb60747dc Mon Sep 17 00:00:00 2001
From: Michael Sippel <micha@fragmental.art>
Date: Wed, 9 Dec 2020 12:51:43 +0100
Subject: [PATCH] ChannelReceiver: add recv() function to receive the complete
 chunk instead of iterating

---
 src/channel.rs | 25 +++++++++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/src/channel.rs b/src/channel.rs
index e6bd5ac..8416416 100644
--- a/src/channel.rs
+++ b/src/channel.rs
@@ -121,6 +121,31 @@ impl<Data: ChannelData> Drop for ChannelSender<Data> {
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
+impl<Data: ChannelData> ChannelReceiver<Data> {
+    pub async fn recv(&self) -> Option<Data> {
+        ChannelRead(self.0.clone()).await
+    }
+}
+
+struct ChannelRead<Data: ChannelData>(Arc<Mutex<ChannelState<Data>>>);
+impl<Data: ChannelData> std::future::Future for ChannelRead<Data> {
+    type Output = Option<Data>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        let mut state = self.0.lock().unwrap();
+        if let Some(buf) = state.send_buf.take() {
+            Poll::Ready(Some(buf))
+        } else if state.num_senders == 0 {
+            Poll::Ready(None)
+        } else {
+            state.waker = Some(cx.waker().clone());
+            Poll::Pending
+        }
+    }
+}
+
+//<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
+
 impl<Data: ChannelData> Stream for ChannelReceiver<Data> {
     type Item = Data::Item;