Fix deadlock again (maybe)

This took like 4 hours to figure out, async sucks wtf

api/rust/src/input.rs

@@ -177,6 +177,7 @@ impl Input {
 
                     while let Some(Ok(_response)) = stream.next().await {
                         action();
+                        tokio::task::yield_now().await;
                     }
                 }
                 .boxed(),
@@ -232,6 +233,7 @@ impl Input {
 
                     while let Some(Ok(_response)) = stream.next().await {
                         action();
+                        tokio::task::yield_now().await;
                     }
                 }
                 .boxed(),

api/rust/src/lib.rs

@@ -82,8 +82,10 @@
 use std::sync::OnceLock;
 
 use futures::{
-    channel::mpsc::UnboundedReceiver, future::BoxFuture, stream::FuturesUnordered, Future,
-    StreamExt,
+    channel::mpsc::UnboundedReceiver,
+    future::{BoxFuture, Either},
+    stream::FuturesUnordered,
+    Future, StreamExt,
 };
 use input::Input;
 use output::Output;
@@ -175,31 +177,17 @@ pub async fn connect(
 ///
 /// This function is inserted at the end of your config through the [`config`] macro.
 /// You should use the macro instead of this function directly.
-pub async fn listen(fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>) {
-    let mut future_set = FuturesUnordered::<
-        BoxFuture<(
-            Option<BoxFuture<()>>,
-            Option<UnboundedReceiver<BoxFuture<()>>>,
-        )>,
-    >::new();
+pub async fn listen(mut fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>) {
+    let mut future_set = FuturesUnordered::<BoxFuture<()>>::new();
 
-    future_set.push(Box::pin(async move {
-        let (fut, stream) = fut_recv.into_future().await;
-        (fut, Some(stream))
-    }));
-
-    while let Some((fut, stream)) = future_set.next().await {
+    loop {
+        match futures::future::select(fut_recv.next(), future_set.next()).await {
+            Either::Left((fut, _)) => {
                 if let Some(fut) = fut {
-            future_set.push(Box::pin(async move {
-                fut.await;
-                (None, None)
-            }));
+                    future_set.push(fut);
                 }
-        if let Some(stream) = stream {
-            future_set.push(Box::pin(async move {
-                let (fut, stream) = stream.into_future().await;
-                (fut, Some(stream))
-            }))
+            }
+            Either::Right(_) => (),
         }
     }
 }

api/rust/src/output.rs

@@ -166,6 +166,7 @@ impl Output {
                         };
 
                         for_all(output);
+                        tokio::task::yield_now().await;
                     }
                 }
                 .boxed(),

api/rust/src/process.rs

@@ -153,6 +153,7 @@ impl Process {
                                 exit(response.exit_code, exit_msg);
                             }
                         }
+                        tokio::task::yield_now().await;
                     }
                 }
                 .boxed(),