Add parallel_stream::any

Author: amadeusine

src/lib.rs

@@ -52,7 +52,7 @@ mod par_stream;
 pub use from_parallel_stream::FromParallelStream;
 pub use from_stream::{from_stream, FromStream};
 pub use into_parallel_stream::IntoParallelStream;
-pub use par_stream::{ForEach, Map, NextFuture, ParallelStream, Take};
+pub use par_stream::{ForEach, Map, NextFuture, ParallelStream, Take, Any};
 
 pub mod prelude;
 pub mod vec;

src/par_stream/any.rs

@@ -0,0 +1,103 @@
+use async_std::prelude::*;
+use async_std::future::Future;
+use pin_project_lite::pin_project;
+use async_std::task::{self, Context, Poll};
+use async_std::sync::{self, Receiver, Sender};
+
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
+
+use crate::ParallelStream;
+
+pin_project! {
+    /// Calls a closure on each element until true or exhausted.
+    #[derive(Debug)]
+    pub struct Any {
+        #[pin]
+        receiver: Receiver<()>,
+        // Track whether the input stream has been exhausted.
+        exhausted: Arc<AtomicBool>,
+        // Count how many tasks are executing.
+        ref_count: Arc<AtomicU64>,
+        // Track the boolean value as executed.
+        value: Arc<AtomicBool>,
+    }
+}
+
+impl Any {
+    /// Creates a new instance of `Any`.
+    pub fn new<S, F, Fut>(mut stream: S, mut f: F) -> Self
+    where
+        S: ParallelStream,
+        F: FnMut(S::Item) -> Fut + Send + Sync + Copy + 'static,
+        Fut: Future<Output = bool> + Send,
+    {
+        let exhausted = Arc::new(AtomicBool::new(false));
+        let value = Arc::new(AtomicBool::new(false));
+        let ref_count = Arc::new(AtomicU64::new(0));
+        let (sender, receiver): (Sender<()>, Receiver<()>) = sync::channel(1);
+        let _limit = stream.get_limit();
+
+        // Initialize the return type here to prevent borrowing issues.
+        let this = Self {
+            receiver,
+            exhausted: exhausted.clone(),
+            ref_count: ref_count.clone(),
+            value: value.clone(),
+        };
+
+        task::spawn(async move {
+            while let Some(item) = stream.next().await {
+                let sender = sender.clone();
+                let exhausted = exhausted.clone();
+                let ref_count = ref_count.clone();
+                let value = value.clone();
+
+                ref_count.fetch_add(1, Ordering::SeqCst);
+
+                task::spawn(async move {
+                    // Execute the closure.
+                    let res = f(item).await;
+
+                    // Wake up the receiver if we know we're done.
+                    ref_count.fetch_sub(1, Ordering::SeqCst);
+                    if res {
+                        value.fetch_or(true, Ordering::SeqCst);
+                        sender.send(()).await;
+                    } else if exhausted.load(Ordering::SeqCst) && ref_count.load(Ordering::SeqCst) == 0 {
+                        sender.send(()).await;
+                    }
+                });
+            }
+
+            // The input stream will no longer yield items.
+            exhausted.store(true, Ordering::SeqCst);
+        });
+
+        this
+    }
+}
+
+impl Future for Any {
+    type Output = bool;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        task::ready!(this.receiver.poll_next(cx));
+        Poll::Ready(this.value.load(Ordering::SeqCst))
+    }
+}
+
+#[async_std::test]
+async fn smoke() {
+    let s = async_std::stream::repeat(5usize);
+    let result = crate::from_stream(s)
+        .take(3)
+        .any(|n| async move {
+            n * 2 < 9
+        })
+        .await;
+    
+    assert_eq!(result, false);
+}

src/par_stream/mod.rs

@@ -9,11 +9,13 @@ pub use for_each::ForEach;
 pub use map::Map;
 pub use next::NextFuture;
 pub use take::Take;
+pub use any::Any;
 
 mod for_each;
 mod map;
 mod next;
 mod take;
+mod any;
 
 /// Parallel version of the standard `Stream` trait.
 pub trait ParallelStream: Sized + Send + Sync + Unpin + 'static {
@@ -63,6 +65,15 @@ pub trait ParallelStream: Sized + Send + Sync + Unpin + 'static {
         ForEach::new(self, f)
     }
 
+    /// Applies `f` to each item of this stream in parallel and returns true if at least one element satisfies `f`.
+    fn any<F, Fut>(self, f: F) -> Any
+    where
+        F: FnMut(Self::Item) -> Fut + Send + Sync + Copy + 'static,
+        Fut: Future<Output = bool> + Send,
+    {
+        Any::new(self, f)
+    }
+
     /// Transforms a stream into a collection.
     ///
     ///`collect()` can take anything streamable, and turn it into a relevant