xmtp_api_backend/queries/stream/
extractor.rs

1//! Convert collections and keep each conversion error in the stream.
2use futures::{Stream, StreamExt, stream};
3
4pub fn try_extractor<S, T, U, E, F>(stream: S, decode: F) -> impl Stream<Item = Result<U, E>>
5where
6    S: Stream<Item = Result<Vec<T>, E>>,
7    F: Fn(T) -> Result<U, E> + Clone,
8{
9    stream
10        .map(move |result| {
11            let items = match result {
12                Ok(items) => items.into_iter().map(decode.clone()).collect(),
13                Err(error) => vec![Err(error)],
14            };
15            stream::iter(items)
16        })
17        .flatten()
18}
19
20#[cfg(test)]
21mod tests {
22    use super::*;
23    #[xmtp_common::test(unwrap_try = true)]
24    async fn preserves_order_and_all_errors() {
25        let input = stream::iter(vec![Ok(vec![]), Ok(vec![1, 2, 3]), Err("wire")]);
26        let output: Vec<_> = try_extractor(input, |n| if n == 2 { Err("decode") } else { Ok(n) })
27            .collect()
28            .await;
29        assert_eq!(output, vec![Ok(1), Err("decode"), Ok(3), Err("wire")]);
30    }
31    #[xmtp_common::test(unwrap_try = true)]
32    async fn empty_stream_finishes() {
33        let output: Vec<_> = try_extractor(stream::empty::<Result<Vec<u32>, ()>>(), Ok)
34            .collect()
35            .await;
36        assert!(output.is_empty());
37    }
38}
xmtp_api_backend/queries/stream/extractor.rs

xmtp_api_backend/queries/stream/
extractor.rs
