Merge pull request #219 from matheus-consoli/look-ma-no-deps

use `FutureGroup` for concurrent streams

Author: yoshuawuyts

Cargo.toml

@@ -29,15 +29,14 @@ harness = false
 [features]
 default = ["std"]
 std = ["alloc", "futures-lite/std"]
-alloc = ["dep:fixedbitset", "dep:smallvec", "dep:futures-buffered", "futures-lite/alloc"]
+alloc = ["dep:fixedbitset", "dep:smallvec", "futures-lite/alloc"]
 
 [dependencies]
 fixedbitset = { version = "0.5.7", default-features = false, optional = true }
 futures-core = { version = "0.3", default-features = false }
 futures-lite = { version = "2.5.0", default-features = false }
 pin-project = "1.1"
 smallvec = { version = "1.13", optional = true }
-futures-buffered = { version = "0.2.9", optional = true }
 
 [dev-dependencies]
 async-io = "2.4"

src/concurrent_stream/for_each.rs

@@ -1,5 +1,5 @@
 use super::{Consumer, ConsumerState};
-use futures_buffered::FuturesUnordered;
+use crate::future::FutureGroup;
 use futures_lite::StreamExt;
 use pin_project::pin_project;
 
@@ -22,7 +22,7 @@ where
     // NOTE: we can remove the `Arc` here if we're willing to make this struct self-referential
     count: Arc<AtomicUsize>,
     #[pin]
-    group: FuturesUnordered<ForEachFut<F, FutT, T, FutB>>,
+    group: FutureGroup<ForEachFut<F, FutT, T, FutB>>,
     limit: usize,
     f: F,
     _phantom: PhantomData<(T, FutB)>,
@@ -44,7 +44,7 @@ where
             f,
             _phantom: PhantomData,
             count: Arc::new(AtomicUsize::new(0)),
-            group: FuturesUnordered::new(),
+            group: FutureGroup::new(),
         }
     }
 }
@@ -69,7 +69,7 @@ where
         // Space was available! - insert the item for posterity
         this.count.fetch_add(1, Ordering::Relaxed);
         let fut = ForEachFut::new(this.f.clone(), future, this.count.clone());
-        this.group.as_mut().push(fut);
+        this.group.as_mut().insert_pinned(fut);
 
         ConsumerState::Continue
     }

src/concurrent_stream/from_concurrent_stream.rs

@@ -1,9 +1,9 @@
 use super::{ConcurrentStream, Consumer, ConsumerState, IntoConcurrentStream};
+use crate::future::FutureGroup;
 #[cfg(all(feature = "alloc", not(feature = "std")))]
 use alloc::vec::Vec;
 use core::future::Future;
 use core::pin::Pin;
-use futures_buffered::FuturesUnordered;
 use futures_lite::StreamExt;
 use pin_project::pin_project;
 
@@ -44,14 +44,14 @@ impl<T, E> FromConcurrentStream<Result<T, E>> for Result<Vec<T>, E> {
 #[pin_project]
 pub(crate) struct VecConsumer<'a, Fut: Future> {
     #[pin]
-    group: FuturesUnordered<Fut>,
+    group: FutureGroup<Fut>,
     output: &'a mut Vec<Fut::Output>,
 }
 
 impl<'a, Fut: Future> VecConsumer<'a, Fut> {
     pub(crate) fn new(output: &'a mut Vec<Fut::Output>) -> Self {
         Self {
-            group: FuturesUnordered::new(),
+            group: FutureGroup::new(),
             output,
         }
     }
@@ -66,7 +66,7 @@ where
     async fn send(self: Pin<&mut Self>, future: Fut) -> super::ConsumerState {
         let mut this = self.project();
         // unbounded concurrency, so we just goooo
-        this.group.as_mut().push(future);
+        this.group.as_mut().insert_pinned(future);
         ConsumerState::Continue
     }
 
@@ -88,14 +88,14 @@ where
 #[pin_project]
 pub(crate) struct ResultVecConsumer<'a, Fut: Future, T, E> {
     #[pin]
-    group: FuturesUnordered<Fut>,
+    group: FutureGroup<Fut>,
     output: &'a mut Result<Vec<T>, E>,
 }
 
 impl<'a, Fut: Future, T, E> ResultVecConsumer<'a, Fut, T, E> {
     pub(crate) fn new(output: &'a mut Result<Vec<T>, E>) -> Self {
         Self {
-            group: FuturesUnordered::new(),
+            group: FutureGroup::new(),
             output,
         }
     }
@@ -110,7 +110,7 @@ where
     async fn send(self: Pin<&mut Self>, future: Fut) -> super::ConsumerState {
         let mut this = self.project();
         // unbounded concurrency, so we just goooo
-        this.group.as_mut().push(future);
+        this.group.as_mut().insert_pinned(future);
         ConsumerState::Continue
     }
 

src/concurrent_stream/try_for_each.rs

@@ -1,6 +1,6 @@
 use crate::concurrent_stream::ConsumerState;
+use crate::future::FutureGroup;
 use crate::private::Try;
-use futures_buffered::FuturesUnordered;
 use futures_lite::StreamExt;
 use pin_project::pin_project;
 
@@ -24,9 +24,8 @@ where
 {
     // NOTE: we can remove the `Arc` here if we're willing to make this struct self-referential
     count: Arc<AtomicUsize>,
-    // TODO: remove the `Pin<Box>` from this signature by requiring this struct is pinned
     #[pin]
-    group: FuturesUnordered<TryForEachFut<F, FutT, T, FutB, B>>,
+    group: FutureGroup<TryForEachFut<F, FutT, T, FutB, B>>,
     limit: usize,
     residual: Option<B::Residual>,
     f: F,
@@ -50,7 +49,7 @@ where
             f,
             residual: None,
             count: Arc::new(AtomicUsize::new(0)),
-            group: FuturesUnordered::new(),
+            group: FutureGroup::new(),
             _phantom: PhantomData,
         }
     }
@@ -93,7 +92,7 @@ where
         // Space was available! - insert the item for posterity
         this.count.fetch_add(1, Ordering::Relaxed);
         let fut = TryForEachFut::new(this.f.clone(), future, this.count.clone());
-        this.group.as_mut().push(fut);
+        this.group.as_mut().insert_pinned(fut);
         ConsumerState::Continue
     }