tuwunel_core/utils/stream/
try_broadband.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 completion-ordered outputs.
9///
10/// Successful item futures may run ahead of downstream demand, and their
11/// results are yielded as they complete. Source errors bypass the transform and
12/// may overtake queued item futures.
13pub trait TryBroadbandExt<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 are
21	/// completion ordered, while source errors bypass `f` and may overtake
22	/// them.
23	fn broadn_and_then<U, F, Fut, N>(
24		self,
25		n: N,
26		f: F,
27	) -> impl TryStream<Ok = U, Error = E, Item = Result<U, E>> + Send
28	where
29		N: Into<Option<usize>>,
30		F: Fn(Self::Ok) -> Fut + Send,
31		Fut: TryFuture<Ok = U, Error = E, Output = Result<U, E>> + Send;
32
33	/// Transforms successful items concurrently with the automatic width.
34	///
35	/// Transformation results are yielded in completion order rather than
36	/// source order. Existing source errors bypass `f` and may overtake queued
37	/// futures.
38	fn broad_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	{
46		self.broadn_and_then(None, f)
47	}
48}
49
50impl<T, E, S> TryBroadbandExt<T, E> for S
51where
52	S: TryStream<Ok = T, Error = E, Item = Result<T, E>> + Send + Sized,
53{
54	fn broadn_and_then<U, F, Fut, N>(
55		self,
56		n: N,
57		f: F,
58	) -> impl TryStream<Ok = U, Error = E, Item = Result<U, E>> + Send
59	where
60		N: Into<Option<usize>>,
61		F: Fn(Self::Ok) -> Fut + Send,
62		Fut: TryFuture<Ok = U, Error = E, Output = Result<U, E>> + Send,
63	{
64		self.map_ok(f)
65			.try_buffer_unordered(n.into().unwrap_or_else(automatic_width))
66	}
67}
tuwunel_core/utils/stream/try_broadband.rs

tuwunel_core/utils/stream/
try_broadband.rs
