add broadband stream extensions

Signed-off-by: Jason Volk <jason@zemos.net>

src/core/utils/stream/broadband.rs

@@ -0,0 +1,84 @@
+//! Broadband stream combinator extensions to futures::Stream
+#![allow(clippy::type_complexity)]
+
+use std::convert::identity;
+
+use futures::{
+	stream::{Stream, StreamExt},
+	Future,
+};
+
+use super::ReadyExt;
+
+const WIDTH: usize = 32;
+
+/// Concurrency extensions to augment futures::StreamExt. broad_ combinators
+/// produce out-of-order
+pub trait BroadbandExt<Item>
+where
+	Self: Stream<Item = Item> + Send + Sized,
+{
+	/// Concurrent filter_map(); unordered results
+	fn broadn_filter_map<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = Option<U>> + Send,
+		U: Send;
+
+	fn broadn_then<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = U> + Send,
+		U: Send;
+
+	#[inline]
+	fn broad_filter_map<F, Fut, U>(self, f: F) -> impl Stream<Item = U> + Send
+	where
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = Option<U>> + Send,
+		U: Send,
+	{
+		self.broadn_filter_map(None, f)
+	}
+
+	#[inline]
+	fn broad_then<F, Fut, U>(self, f: F) -> impl Stream<Item = U> + Send
+	where
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = U> + Send,
+		U: Send,
+	{
+		self.broadn_then(None, f)
+	}
+}
+
+impl<Item, S> BroadbandExt<Item> for S
+where
+	S: Stream<Item = Item> + Send + Sized,
+{
+	#[inline]
+	fn broadn_filter_map<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = Option<U>> + Send,
+		U: Send,
+	{
+		self.map(f)
+			.buffer_unordered(n.into().unwrap_or(WIDTH))
+			.ready_filter_map(identity)
+	}
+
+	#[inline]
+	fn broadn_then<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = U> + Send,
+		U: Send,
+	{
+		self.map(f).buffer_unordered(n.into().unwrap_or(WIDTH))
+	}
+}

src/core/utils/stream/mod.rs

@@ -1,3 +1,4 @@
+mod broadband;
 mod cloned;
 mod expect;
 mod ignore;
@@ -5,7 +6,9 @@ mod iter_stream;
 mod ready;
 mod tools;
 mod try_ready;
+mod wideband;
 
+pub use broadband::BroadbandExt;
 pub use cloned::Cloned;
 pub use expect::TryExpect;
 pub use ignore::TryIgnore;
@@ -13,3 +16,4 @@ pub use iter_stream::IterStream;
 pub use ready::ReadyExt;
 pub use tools::Tools;
 pub use try_ready::TryReadyExt;
+pub use wideband::WidebandExt;

src/core/utils/stream/wideband.rs

@@ -0,0 +1,84 @@
+//! Wideband stream combinator extensions to futures::Stream
+#![allow(clippy::type_complexity)]
+
+use std::convert::identity;
+
+use futures::{
+	stream::{Stream, StreamExt},
+	Future,
+};
+
+use super::ReadyExt;
+
+const WIDTH: usize = 32;
+
+/// Concurrency extensions to augment futures::StreamExt. wideband_ combinators
+/// produce in-order.
+pub trait WidebandExt<Item>
+where
+	Self: Stream<Item = Item> + Send + Sized,
+{
+	/// Concurrent filter_map(); ordered results
+	fn widen_filter_map<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = Option<U>> + Send,
+		U: Send;
+
+	fn widen_then<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = U> + Send,
+		U: Send;
+
+	#[inline]
+	fn wide_filter_map<F, Fut, U>(self, f: F) -> impl Stream<Item = U> + Send
+	where
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = Option<U>> + Send,
+		U: Send,
+	{
+		self.widen_filter_map(None, f)
+	}
+
+	#[inline]
+	fn wide_then<F, Fut, U>(self, f: F) -> impl Stream<Item = U> + Send
+	where
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = U> + Send,
+		U: Send,
+	{
+		self.widen_then(None, f)
+	}
+}
+
+impl<Item, S> WidebandExt<Item> for S
+where
+	S: Stream<Item = Item> + Send + Sized,
+{
+	#[inline]
+	fn widen_filter_map<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = Option<U>> + Send,
+		U: Send,
+	{
+		self.map(f)
+			.buffered(n.into().unwrap_or(WIDTH))
+			.ready_filter_map(identity)
+	}
+
+	#[inline]
+	fn widen_then<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
+	where
+		N: Into<Option<usize>>,
+		F: Fn(Item) -> Fut + Send,
+		Fut: Future<Output = U> + Send,
+		U: Send,
+	{
+		self.map(f).buffered(n.into().unwrap_or(WIDTH))
+	}
+}