sse_parse/
stream.rs

1use alloc::string::String;
2use core::fmt;
3use core::pin::Pin;
4use core::task::{Context, Poll};
5use core::time::Duration;
6
7use futures_core::stream::{FusedStream, Stream};
8
9use crate::{Event, SseParser};
10
11/// Error item of an [`EventStream`].
12///
13/// Every byte sequence is a valid event stream, and invalid UTF-8 decodes
14/// to U+FFFD, so parsing produces no error. The only error an event stream
15/// can carry is one from the underlying byte stream.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub enum Error<E> {
18    /// The byte stream yielded an error.
19    Transport(E),
20}
21
22impl<E: fmt::Display> fmt::Display for Error<E> {
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        match self {
25            Error::Transport(inner) => write!(f, "transport error: {inner}"),
26        }
27    }
28}
29
30#[cfg(feature = "std")]
31impl<E: std::error::Error + 'static> std::error::Error for Error<E> {
32    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
33        match self {
34            Error::Transport(inner) => Some(inner),
35        }
36    }
37}
38
39pin_project_lite::pin_project! {
40    /// Adapter from a stream of byte chunks to a stream of SSE events.
41    ///
42    /// Works with any `Stream<Item = Result<B, E>>` where `B: AsRef<[u8]>`,
43    /// which covers `Vec<u8>`, `&[u8]`, `bytes::Bytes`, and HTTP client
44    /// bodies. Executor-agnostic: the only async dependency is the
45    /// `futures-core` `Stream` trait.
46    ///
47    /// Every completed event is yielded before the inner stream is polled
48    /// again. When the inner stream ends, the parser is finished, remaining
49    /// events are yielded, and the stream terminates. Polling after
50    /// termination returns `Ready(None)` without touching the inner stream.
51    /// A transport error is yielded as an item and the stream stays usable.
52    pub struct EventStream<S> {
53        #[pin]
54        inner: S,
55        parser: SseParser,
56        done: bool,
57    }
58}
59
60impl<S> EventStream<S> {
61    /// Wrap a byte-chunk stream.
62    pub fn new(stream: S) -> Self {
63        EventStream {
64            inner: stream,
65            parser: SseParser::new(),
66            done: false,
67        }
68    }
69
70    /// Seed the last event ID string, for reconnection. Events dispatched
71    /// before the stream's first `id` field report this ID.
72    pub fn set_last_event_id(&mut self, id: impl Into<String>) {
73        self.parser.set_last_event_id(id);
74    }
75
76    /// The last event ID string. See [`SseParser::last_event_id`].
77    pub fn last_event_id(&self) -> &str {
78        self.parser.last_event_id()
79    }
80
81    /// Reconnection time from the most recent valid `retry` field, if any.
82    /// See [`SseParser::retry`].
83    pub fn retry(&self) -> Option<Duration> {
84        self.parser.retry()
85    }
86}
87
88impl<S: fmt::Debug> fmt::Debug for EventStream<S> {
89    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
90        f.debug_struct("EventStream")
91            .field("inner", &self.inner)
92            .field("done", &self.done)
93            .field("last_event_id", &self.parser.last_event_id())
94            .field("retry", &self.parser.retry())
95            .field("buffered_bytes", &self.parser.stats().buffered_bytes)
96            .finish_non_exhaustive()
97    }
98}
99
100impl<S, B, E> Stream for EventStream<S>
101where
102    S: Stream<Item = Result<B, E>>,
103    B: AsRef<[u8]>,
104{
105    type Item = Result<Event, Error<E>>;
106
107    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
108        let mut this = self.project();
109        loop {
110            if let Some(event) = this.parser.next_event() {
111                return Poll::Ready(Some(Ok(event)));
112            }
113            if *this.done {
114                return Poll::Ready(None);
115            }
116            match this.inner.as_mut().poll_next(cx) {
117                Poll::Pending => return Poll::Pending,
118                Poll::Ready(Some(Ok(chunk))) => this.parser.feed(chunk.as_ref()),
119                Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(Error::Transport(e)))),
120                Poll::Ready(None) => {
121                    *this.done = true;
122                    this.parser.finish();
123                }
124            }
125        }
126    }
127}
128
129impl<S, B, E> FusedStream for EventStream<S>
130where
131    S: Stream<Item = Result<B, E>>,
132    B: AsRef<[u8]>,
133{
134    fn is_terminated(&self) -> bool {
135        self.done && !self.parser.has_pending_events()
136    }
137}
sse_parse/stream.rs

sse_parse/
stream.rs
