Upgrade rust/crates/tokio-test to 0.4.1 am: 9dcb441be3 am: 73366b41de am: 1c909e8b5c am: 85ead0ebbb

Original change: https://android-review.googlesource.com/c/platform/external/rust/crates/tokio-test/+/1663104

Change-Id: I06ba81828b6f5ca9e18e2a4791cffe376dca9331

1 files changed, 4 insertions, 8 deletions

diff --git a/src/io.rs b/src/io.rs
index 77adfc3..4ec66a4 100644
--- a/src/io.rs
+++ b/src/io.rs
@@ -21,6 +21,7 @@
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::mpsc;
 use tokio::time::{self, Duration, Instant, Sleep};
+use tokio_stream::wrappers::UnboundedReceiverStream;
 
 use futures_core::{ready, Stream};
 use std::collections::VecDeque;
@@ -69,8 +70,7 @@ struct Inner {
     waiting: Option<Instant>,
     sleep: Option<Pin<Box<Sleep>>>,
     read_wait: Option<Waker>,
-    // rx: mpsc::UnboundedReceiver<Action>,
-    rx: Pin<Box<dyn Stream<Item = Action> + Send>>,
+    rx: UnboundedReceiverStream<Action>,
 }
 
 impl Builder {
@@ -185,13 +185,9 @@ impl Handle {
 
 impl Inner {
     fn new(actions: VecDeque<Action>) -> (Inner, Handle) {
-        let (tx, mut rx) = mpsc::unbounded_channel();
+        let (tx, rx) = mpsc::unbounded_channel();
 
-        let rx = Box::pin(async_stream::stream! {
-            while let Some(item) = rx.recv().await {
-                yield item;
-            }
-        });
+        let rx = UnboundedReceiverStream::new(rx);
 
         let inner = Inner {
             actions,