rx_rust/operators/others/
observable_stream.rs

1use crate::{
2    observable::Observable,
3    operators::others::observable_try_stream::{ObservableTryStream, StreamBuffer, Unbounded},
4    utils::types::MaybeSend,
5};
6use educe::Educe;
7use futures::Stream;
8use std::{convert::Infallible, task::Poll};
9
10/// Converts an Observable that cannot fail into a `futures::Stream` that can be used with
11/// `async/await`.
12///
13/// Each item is yielded as is and completion ends the stream. This is [`ObservableTryStream`]
14/// without the error it can never carry; see there for how it subscribes and buffers.
15///
16/// # Examples
17/// ```rust
18/// use futures::StreamExt;
19/// use rx_rust::{
20///     operators::{
21///         creating::from_iter::FromIter,
22///         others::observable_stream::ObservableStream,
23///     },
24/// };
25///
26/// futures::executor::block_on(async {
27///     let source = FromIter::new(vec![1, 2, 3]);
28///     let mut stream = ObservableStream::new(source);
29///     let values: Vec<_> = (&mut stream).collect().await;
30///     assert_eq!(values, vec![1, 2, 3]);
31/// });
32/// ```
33#[derive(Educe)]
34#[educe(Debug)]
35pub struct ObservableStream<'or, T, OE, B = Unbounded<T>>
36where
37    OE: Observable<'or, T, Infallible>,
38{
39    stream: ObservableTryStream<'or, T, Infallible, OE, B>,
40}
41
42impl<'or, T, OE> ObservableStream<'or, T, OE>
43where
44    OE: Observable<'or, T, Infallible>,
45{
46    /// Buffers every item until it is polled; see [`Unbounded`].
47    pub fn new(source: OE) -> Self {
48        Self::with_buffer(source, Unbounded::new())
49    }
50}
51
52impl<'or, T, OE, B> ObservableStream<'or, T, OE, B>
53where
54    OE: Observable<'or, T, Infallible>,
55    B: StreamBuffer<T>,
56{
57    /// Keeps the items that arrive between two polls in `buffer`; see
58    /// [`ObservableTryStream::with_buffer`].
59    pub fn with_buffer(source: OE, buffer: B) -> Self {
60        Self {
61            stream: ObservableTryStream::with_buffer(source, buffer),
62        }
63    }
64}
65
66impl<'or, T, OE, B> Unpin for ObservableStream<'or, T, OE, B> where
67    OE: Observable<'or, T, Infallible>
68{
69}
70
71impl<'or, T, OE, B> Stream for ObservableStream<'or, T, OE, B>
72where
73    T: MaybeSend + 'or,
74    OE: Observable<'or, T, Infallible>,
75    B: StreamBuffer<T> + MaybeSend + 'or,
76{
77    type Item = B::Item;
78
79    fn poll_next(
80        mut self: std::pin::Pin<&mut Self>,
81        cx: &mut std::task::Context<'_>,
82    ) -> Poll<Option<Self::Item>> {
83        std::pin::Pin::new(&mut self.stream)
84            .poll_next(cx)
85            .map(|item| {
86                item.map(|result| {
87                    let Ok(value) = result;
88                    value
89                })
90            })
91    }
92}
rx_rust/operators/others/observable_stream.rs

rx_rust/operators/others/
observable_stream.rs
