tuwunel_core/utils/stream/
try_wideband.rs

1//! Synchronous combinator extensions to futures::TryStream
2
3use futures::{TryFuture, TryStream, TryStreamExt};
4
5use super::automatic_width;
6use crate::Result;
7
8/// Adds bounded concurrent transformations with ordered transform results.
9///
10/// Successful item futures may run ahead of downstream demand, and their
11/// results retain queue order. Source errors are propagated immediately and may
12/// overtake queued transformation futures.
13pub trait TryWidebandExt<T, E>
14where
15	Self: TryStream<Ok = T, Error = E, Item = Result<T, E>> + Send + Sized,
16{
17	/// Transforms successful 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. Transformation results
21	/// retain queue order, while source errors may overtake them.
22	fn widen_and_then<U, F, Fut, N>(
23		self,
24		n: N,
25		f: F,
26	) -> impl TryStream<Ok = U, Error = E, Item = Result<U, E>> + Send
27	where
28		N: Into<Option<usize>>,
29		F: Fn(Self::Ok) -> Fut + Send,
30		Fut: TryFuture<Ok = U, Error = E, Output = Result<U, E>> + Send,
31		U: Send;
32
33	/// Transforms successful items concurrently with the automatic width.
34	///
35	/// Item futures may run ahead, but transformation results retain queue
36	/// order. Existing source errors bypass `f` and may overtake queued
37	/// futures.
38	fn wide_and_then<U, F, Fut>(
39		self,
40		f: F,
41	) -> impl TryStream<Ok = U, Error = E, Item = Result<U, E>> + Send
42	where
43		F: Fn(Self::Ok) -> Fut + Send,
44		Fut: TryFuture<Ok = U, Error = E, Output = Result<U, E>> + Send,
45		U: Send,
46	{
47		self.widen_and_then(None, f)
48	}
49}
50
51impl<T, E, S> TryWidebandExt<T, E> for S
52where
53	S: TryStream<Ok = T, Error = E, Item = Result<T, E>> + Send + Sized,
54	E: Send,
55{
56	fn widen_and_then<U, F, Fut, N>(
57		self,
58		n: N,
59		f: F,
60	) -> impl TryStream<Ok = U, Error = E, Item = Result<U, E>> + Send
61	where
62		N: Into<Option<usize>>,
63		F: Fn(Self::Ok) -> Fut + Send,
64		Fut: TryFuture<Ok = U, Error = E, Output = Result<U, E>> + Send,
65		U: Send,
66	{
67		self.map_ok(f)
68			.try_buffered(n.into().unwrap_or_else(automatic_width))
69	}
70}
tuwunel_core/utils/stream/try_wideband.rs

tuwunel_core/utils/stream/
try_wideband.rs
