chore: Use tokio-stream UnboundedReceiverStream (#831)

Cargo.toml

@@ -20,7 +20,7 @@ rand = "0.9"
 slab = "0.4.9"
 sync_wrapper = "1"
 tokio = "1.6.2"
-tokio-stream = "0.1.0"
+tokio-stream = "0.1.1"
 tokio-test = "0.4"
 tokio-util = { version = "0.7.0", default-features = false }
 tracing = { version = "0.1.2", default-features = false }

tower/Cargo.toml

@@ -76,7 +76,6 @@ sync_wrapper = { workspace = true, optional = true }
 [dev-dependencies]
 futures = { workspace = true }
 hdrhistogram = { workspace = true }
-pin-project-lite = { workspace = true }
 tokio = { workspace = true, features = ["macros", "sync", "test-util", "rt-multi-thread"] }
 tokio-stream = { workspace = true }
 tokio-test = { workspace = true }

tower/tests/balance/main.rs

@@ -4,6 +4,7 @@ mod support;
 
 use std::future::Future;
 use std::task::{Context, Poll};
+use tokio_stream::wrappers::UnboundedReceiverStream;
 use tokio_test::{assert_pending, assert_ready, task};
 use tower::balance::p2c::Balance;
 use tower::discover::Change;
@@ -37,7 +38,7 @@ fn stress() {
     let _t = support::trace_init();
     let mut task = task::spawn(());
     let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Result<_, &'static str>>();
-    let mut cache = Balance::<_, Req>::new(support::IntoStream::new(rx));
+    let mut cache = Balance::<_, Req>::new(UnboundedReceiverStream::new(rx));
 
     let mut nready = 0;
     let mut services = slab::Slab::<(mock::Handle<Req, Req>, bool)>::new();

tower/tests/support.rs

@@ -2,10 +2,7 @@
 
 use std::fmt;
 use std::future;
-use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::sync::mpsc;
-use tokio_stream::Stream;
 use tower::Service;
 
 pub(crate) fn trace_init() -> tracing::subscriber::DefaultGuard {
@@ -17,36 +14,6 @@ pub(crate) fn trace_init() -> tracing::subscriber::DefaultGuard {
     tracing::subscriber::set_default(subscriber)
 }
 
-pin_project_lite::pin_project! {
-    #[derive(Clone, Debug)]
-    pub struct IntoStream<S> {
-        #[pin]
-        inner: S
-    }
-}
-
-impl<S> IntoStream<S> {
-    pub fn new(inner: S) -> Self {
-        Self { inner }
-    }
-}
-
-impl<I> Stream for IntoStream<mpsc::Receiver<I>> {
-    type Item = I;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project().inner.poll_recv(cx)
-    }
-}
-
-impl<I> Stream for IntoStream<mpsc::UnboundedReceiver<I>> {
-    type Item = I;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project().inner.poll_recv(cx)
-    }
-}
-
 #[derive(Clone, Debug)]
 pub struct AssertSpanSvc {
     span: tracing::Span,

tower/tests/util/call_all.rs

@@ -5,6 +5,7 @@ use std::fmt;
 use std::future::{ready, Future, Ready};
 use std::task::{Context, Poll};
 use std::{cell::Cell, rc::Rc};
+use tokio_stream::wrappers::UnboundedReceiverStream;
 use tokio_test::{assert_pending, assert_ready, task};
 use tower::util::ServiceExt;
 use tower_service::*;
@@ -50,7 +51,7 @@ fn ordered() {
         admit: admit.clone(),
     };
     let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
-    let ca = srv.call_all(support::IntoStream::new(rx));
+    let ca = srv.call_all(UnboundedReceiverStream::new(rx));
     pin_mut!(ca);
 
     assert_pending!(mock.enter(|cx, _| ca.as_mut().poll_next(cx)));
@@ -152,7 +153,7 @@ async fn pending() {
     let mut task = task::spawn(());
 
     let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
-    let ca = mock.call_all(support::IntoStream::new(rx));
+    let ca = mock.call_all(UnboundedReceiverStream::new(rx));
     pin_mut!(ca);
 
     assert_pending!(task.enter(|cx, _| ca.as_mut().poll_next(cx)));