tuwunel_core/utils/stream/
wideband.rs

1//! Wideband stream combinator extensions to futures::Stream
2
3use std::convert::identity;
4
5use futures::stream::{Stream, StreamExt};
6
7use super::{ReadyExt, automatic_width};
8
9/// Adds bounded concurrent transformations that preserve stream order.
10///
11/// Multiple item futures may run ahead of downstream demand. Completed outputs
12/// are held until every earlier input has produced its output.
13pub trait WidebandExt<Item>
14where
15	Self: Stream<Item = Item> + Send + Sized,
16{
17	/// Maps and filters items concurrently with an explicit width.
18	///
19	/// `n` limits in-flight item futures, while `None` selects the automatic
20	/// width; an explicit zero cannot make progress. Present outputs retain
21	/// source order and absent outputs are omitted.
22	fn widen_filter_map<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
23	where
24		N: Into<Option<usize>>,
25		F: Fn(Item) -> Fut + Send,
26		Fut: Future<Output = Option<U>> + Send,
27		U: Send;
28
29	/// Maps items concurrently with an explicit width while preserving order.
30	///
31	/// `n` limits in-flight item futures, while `None` selects the automatic
32	/// width; an explicit zero cannot make progress. Item futures may run
33	/// ahead, but outputs retain source order.
34	fn widen_then<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
35	where
36		N: Into<Option<usize>>,
37		F: Fn(Item) -> Fut + Send,
38		Fut: Future<Output = U> + Send,
39		U: Send;
40
41	/// Maps and filters items concurrently with the automatic width.
42	///
43	/// Present outputs retain source order even when their futures complete out
44	/// of order. Absent outputs are omitted.
45	#[inline]
46	fn wide_filter_map<F, Fut, U>(self, f: F) -> impl Stream<Item = U> + Send
47	where
48		F: Fn(Item) -> Fut + Send,
49		Fut: Future<Output = Option<U>> + Send,
50		U: Send,
51	{
52		self.widen_filter_map(None, f)
53	}
54
55	/// Maps items concurrently with the automatic width while preserving order.
56	///
57	/// Item futures may run ahead of downstream demand. Completed outputs wait
58	/// for every earlier input before being yielded.
59	#[inline]
60	fn wide_then<F, Fut, U>(self, f: F) -> impl Stream<Item = U> + Send
61	where
62		F: Fn(Item) -> Fut + Send,
63		Fut: Future<Output = U> + Send,
64		U: Send,
65	{
66		self.widen_then(None, f)
67	}
68}
69
70impl<Item, S> WidebandExt<Item> for S
71where
72	S: Stream<Item = Item> + Send + Sized,
73{
74	#[inline]
75	fn widen_filter_map<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
76	where
77		N: Into<Option<usize>>,
78		F: Fn(Item) -> Fut + Send,
79		Fut: Future<Output = Option<U>> + Send,
80		U: Send,
81	{
82		self.map(f)
83			.buffered(n.into().unwrap_or_else(automatic_width))
84			.ready_filter_map(identity)
85	}
86
87	#[inline]
88	fn widen_then<F, Fut, U, N>(self, n: N, f: F) -> impl Stream<Item = U> + Send
89	where
90		N: Into<Option<usize>>,
91		F: Fn(Item) -> Fut + Send,
92		Fut: Future<Output = U> + Send,
93		U: Send,
94	{
95		self.map(f)
96			.buffered(n.into().unwrap_or_else(automatic_width))
97	}
98}
tuwunel_core/utils/stream/wideband.rs

tuwunel_core/utils/stream/
wideband.rs
