tuwunel_core/utils/stream/expect.rs
1use futures::{Stream, StreamExt, TryStream};
2
3use crate::Result;
4
5/// Converts fallible stream items into values by expecting success.
6///
7/// Successful items pass through in source order. Errors terminate the current
8/// poll by panicking with either a default or caller-supplied message.
9pub trait TryExpect<Item>
10where
11 Item: Send,
12 Self: Send + Sized,
13{
14 /// Expects every stream item with the default failure message.
15 ///
16 /// Successful values are unwrapped lazily as the stream is polled. The
17 /// default message identifies a stream expectation failure.
18 ///
19 /// # Panics
20 ///
21 /// Panics when the source stream yields an error.
22 fn expect_ok(self) -> impl Stream<Item = Item> + Send;
23
24 /// Expects every stream item with `msg` as the failure message.
25 ///
26 /// Successful values are unwrapped lazily as the stream is polled. The
27 /// supplied message is used for every failing item.
28 ///
29 /// # Panics
30 ///
31 /// Panics when the source stream yields an error.
32 fn map_expect(self, msg: &str) -> impl Stream<Item = Item> + Send;
33}
34
35impl<Item, S> TryExpect<Item> for S
36where
37 S: Stream<Item = Result<Item>> + Send + TryStream,
38 Item: Send,
39 Self: Send + Sized,
40{
41 #[inline]
42 fn expect_ok(self: S) -> impl Stream<Item = Item> + Send {
43 self.map_expect("stream expectation failure")
44 }
45
46 //TODO: move to impl MapExpect
47 #[inline]
48 fn map_expect(self, msg: &str) -> impl Stream<Item = Item> + Send {
49 self.map(|res| res.expect(msg))
50 }
51}