diff options
| -rw-r--r-- | src/recv.rs | 18 |
1 files changed, 8 insertions, 10 deletions
diff --git a/src/recv.rs b/src/recv.rs index 4d1047b..3bea709 100644 --- a/src/recv.rs +++ b/src/recv.rs @@ -15,16 +15,16 @@ use crate::stream::*; /// Structure to warn when the sender is dropped before end of stream was reached, like when /// connection to some remote drops while transmitting data struct Sender { - inner: Option<mpsc::Sender<Packet>>, + inner: Option<mpsc::UnboundedSender<Packet>>, } impl Sender { - fn new(inner: mpsc::Sender<Packet>) -> Self { + fn new(inner: mpsc::UnboundedSender<Packet>) -> Self { Sender { inner: Some(inner) } } - async fn send(&self, packet: Packet) { - let _ = self.inner.as_ref().unwrap().send(packet).await; + fn send(&self, packet: Packet) { + let _ = self.inner.as_ref().unwrap().send(packet); } fn end(&mut self) { @@ -35,9 +35,7 @@ impl Sender { impl Drop for Sender { fn drop(&mut self) { if let Some(inner) = self.inner.take() { - tokio::spawn(async move { - let _ = inner.send(Err(255)).await; - }); + let _ = inner.send(Err(255)); } } } @@ -102,18 +100,18 @@ pub(crate) trait RecvLoop: Sync + 'static { let mut sender = if let Some(send) = streams.remove(&(id)) { send } else { - let (send, recv) = mpsc::channel(4); + let (send, recv) = mpsc::unbounded_channel(); trace!("recv_loop: id {} is new channel", id); self.recv_handler( id, - Box::pin(tokio_stream::wrappers::ReceiverStream::new(recv)), + Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(recv)), ); Sender::new(send) }; // If we get an error, the receiving end is disconnected. // We still need to reach eos before dropping this sender - let _ = sender.send(packet).await; + let _ = sender.send(packet); if has_cont { assert!(!is_error); |