Add map_stream(), trait constraints to OptionExt.

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

src/core/utils/mod.rs

@@ -33,6 +33,7 @@ pub use self::{
 	hash::sha256::delimited as calculate_hash,
 	json::{deserialize_from_str, to_canonical_object},
 	mutex_map::{Guard as MutexMapGuard, MutexMap},
+	option::OptionExt,
 	rand::{shuffle, string as random_string},
 	stream::{IterStream, ReadyExt, Tools as StreamTools, TryReadyExt},
 	string::{str_from_bytes, string_from_bytes},

src/core/utils/option.rs

@@ -1,11 +1,39 @@
-use futures::future::OptionFuture;
+use futures::{FutureExt, Stream, future::OptionFuture};
 
-pub trait OptionExt<T> {
+use super::IterStream;
-	fn map_async<O: Future, F: FnOnce(T) -> O>(self, f: F) -> OptionFuture<O>;
-}
 
-impl<T> OptionExt<T> for Option<T> {
+pub trait OptionExt<Fut, T, U>
-	fn map_async<O: Future, F: FnOnce(T) -> O>(self, f: F) -> OptionFuture<O> {
+where
-		OptionFuture::<_>::from(self.map(f))
+	Fut: Future<Output = U> + Send,
+	U: Send,
+{
+	fn map_async<F>(self, f: F) -> OptionFuture<Fut>
+	where
+		F: FnOnce(T) -> Fut;
+
+	#[inline]
+	fn map_stream<F>(self, f: F) -> impl Stream<Item = U> + Send
+	where
+		F: FnOnce(T) -> Fut,
+		Self: Sized,
+	{
+		self.map_async(f)
+			.map(Option::into_iter)
+			.map(IterStream::stream)
+			.flatten_stream()
+	}
+}
+
+impl<Fut, T, U> OptionExt<Fut, T, U> for Option<T>
+where
+	Fut: Future<Output = U> + Send,
+	U: Send,
+{
+	#[inline]
+	fn map_async<F>(self, f: F) -> OptionFuture<Fut>
+	where
+		F: FnOnce(T) -> Fut,
+	{
+		self.map(f).into()
 	}
 }