sse-parse 0.1.0

use alloc::string::String;
use core::fmt;
use core::pin::Pin;
use core::task::{Context, Poll};
use core::time::Duration;

use futures_core::stream::{FusedStream, Stream};

use crate::{Event, SseParser};

/// Error item of an [`EventStream`].
///
/// Every byte sequence is a valid event stream, and invalid UTF-8 decodes
/// to U+FFFD, so parsing produces no error. The only error an event stream
/// can carry is one from the underlying byte stream.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Error<E> {
    /// The byte stream yielded an error.
    Transport(E),
}

impl<E: fmt::Display> fmt::Display for Error<E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Error::Transport(inner) => write!(f, "transport error: {inner}"),
        }
    }
}

#[cfg(feature = "std")]
impl<E: std::error::Error + 'static> std::error::Error for Error<E> {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Error::Transport(inner) => Some(inner),
        }
    }
}

pin_project_lite::pin_project! {
    /// Adapter from a stream of byte chunks to a stream of SSE events.
    ///
    /// Works with any `Stream<Item = Result<B, E>>` where `B: AsRef<[u8]>`,
    /// which covers `Vec<u8>`, `&[u8]`, `bytes::Bytes`, and HTTP client
    /// bodies. Executor-agnostic: the only async dependency is the
    /// `futures-core` `Stream` trait.
    ///
    /// Every completed event is yielded before the inner stream is polled
    /// again. When the inner stream ends, the parser is finished, remaining
    /// events are yielded, and the stream terminates. Polling after
    /// termination returns `Ready(None)` without touching the inner stream.
    /// A transport error is yielded as an item and the stream stays usable.
    pub struct EventStream<S> {
        #[pin]
        inner: S,
        parser: SseParser,
        done: bool,
    }
}

impl<S> EventStream<S> {
    /// Wrap a byte-chunk stream.
    pub fn new(stream: S) -> Self {
        EventStream {
            inner: stream,
            parser: SseParser::new(),
            done: false,
        }
    }

    /// Seed the last event ID string, for reconnection. Events dispatched
    /// before the stream's first `id` field report this ID.
    pub fn set_last_event_id(&mut self, id: impl Into<String>) {
        self.parser.set_last_event_id(id);
    }

    /// The last event ID string. See [`SseParser::last_event_id`].
    pub fn last_event_id(&self) -> &str {
        self.parser.last_event_id()
    }

    /// Reconnection time from the most recent valid `retry` field, if any.
    /// See [`SseParser::retry`].
    pub fn retry(&self) -> Option<Duration> {
        self.parser.retry()
    }
}

impl<S: fmt::Debug> fmt::Debug for EventStream<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EventStream")
            .field("inner", &self.inner)
            .field("done", &self.done)
            .field("last_event_id", &self.parser.last_event_id())
            .field("retry", &self.parser.retry())
            .field("buffered_bytes", &self.parser.stats().buffered_bytes)
            .finish_non_exhaustive()
    }
}

impl<S, B, E> Stream for EventStream<S>
where
    S: Stream<Item = Result<B, E>>,
    B: AsRef<[u8]>,
{
    type Item = Result<Event, Error<E>>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let mut this = self.project();
        loop {
            if let Some(event) = this.parser.next_event() {
                return Poll::Ready(Some(Ok(event)));
            }
            if *this.done {
                return Poll::Ready(None);
            }
            match this.inner.as_mut().poll_next(cx) {
                Poll::Pending => return Poll::Pending,
                Poll::Ready(Some(Ok(chunk))) => this.parser.feed(chunk.as_ref()),
                Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(Error::Transport(e)))),
                Poll::Ready(None) => {
                    *this.done = true;
                    this.parser.finish();
                }
            }
        }
    }
}

impl<S, B, E> FusedStream for EventStream<S>
where
    S: Stream<Item = Result<B, E>>,
    B: AsRef<[u8]>,
{
    fn is_terminated(&self) -> bool {
        self.done && !self.parser.has_pending_events()
    }
}