sse-parse 0.1.0

//! The `EventStream` adapter's poll, error, and termination contract.

mod common;

use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use common::{collect_stream, noop_waker, ScriptedStream, Step};
use futures_core::stream::{FusedStream, Stream};
use sse_parse::{Error, Event, EventStream};

type Script = Vec<Step<Vec<u8>, String>>;

fn events(items: Vec<Result<Event, Error<String>>>) -> Vec<Result<Event, String>> {
    items
        .into_iter()
        .map(|r| r.map_err(|Error::Transport(e)| e))
        .collect()
}

#[test]
fn yields_all_completed_events_before_polling_inner_again() {
    let script: Script = vec![
        Step::Chunk(b"data: a\n\ndata: b\n\ndata: c\n\n".to_vec()),
        Step::Chunk(b"data: d\n\n".to_vec()),
    ];
    let inner = ScriptedStream::new(script);
    let mut stream = EventStream::new(inner);
    let waker = noop_waker();
    let mut cx = Context::from_waker(&waker);
    let mut datas = Vec::new();
    for _ in 0..3 {
        match Pin::new(&mut stream).poll_next(&mut cx) {
            Poll::Ready(Some(Ok(ev))) => datas.push(ev.data),
            other => panic!("unexpected {other:?}"),
        }
    }
    assert_eq!(datas, ["a", "b", "c"]);
}

#[test]
fn pending_propagates_and_resumes() {
    let script: Script = vec![
        Step::Pending,
        Step::Chunk(b"data: a".to_vec()),
        Step::Pending,
        Step::Chunk(b"\n\n".to_vec()),
        Step::Pending,
        Step::End,
    ];
    let mut stream = EventStream::new(ScriptedStream::new(script));
    let waker = noop_waker();
    let mut cx = Context::from_waker(&waker);
    assert!(matches!(
        Pin::new(&mut stream).poll_next(&mut cx),
        Poll::Pending
    ));
    // Second poll consumes "data: a" (no event) and then hits Pending.
    assert!(matches!(
        Pin::new(&mut stream).poll_next(&mut cx),
        Poll::Pending
    ));
    match Pin::new(&mut stream).poll_next(&mut cx) {
        Poll::Ready(Some(Ok(ev))) => assert_eq!(ev.data, "a"),
        other => panic!("unexpected {other:?}"),
    }
    assert!(matches!(
        Pin::new(&mut stream).poll_next(&mut cx),
        Poll::Pending
    ));
    assert!(matches!(
        Pin::new(&mut stream).poll_next(&mut cx),
        Poll::Ready(None)
    ));
}

#[test]
fn inner_end_finishes_the_parser_and_fuses() {
    let script: Script = vec![
        Step::Chunk(b"data: a\rdata: b\r\r".to_vec()),
        Step::Chunk(b"data: lost\n".to_vec()),
        Step::End,
        // Anything after End must never be observed.
        Step::Chunk(b"data: never\n\n".to_vec()),
    ];
    let inner = ScriptedStream::new(script);
    let mut stream = EventStream::new(inner);
    let waker = noop_waker();
    let mut cx = Context::from_waker(&waker);
    assert!(!stream.is_terminated());
    match Pin::new(&mut stream).poll_next(&mut cx) {
        Poll::Ready(Some(Ok(ev))) => assert_eq!(ev.data, "a\nb"),
        other => panic!("unexpected {other:?}"),
    }
    assert!(matches!(
        Pin::new(&mut stream).poll_next(&mut cx),
        Poll::Ready(None)
    ));
    assert!(stream.is_terminated());
    for _ in 0..3 {
        assert!(matches!(
            Pin::new(&mut stream).poll_next(&mut cx),
            Poll::Ready(None)
        ));
    }
}

#[test]
fn transport_error_is_yielded_and_stream_stays_usable() {
    let script: Script = vec![
        Step::Chunk(b"data: a\n\ndata: b".to_vec()),
        Step::Fail("boom".to_string()),
        Step::Chunk(b"\n\n".to_vec()),
        Step::End,
    ];
    let items = events(collect_stream(EventStream::new(ScriptedStream::new(
        script,
    ))));
    assert_eq!(items.len(), 3);
    assert_eq!(items[0].as_ref().unwrap().data, "a");
    assert_eq!(items[1], Err("boom".to_string()));
    assert_eq!(items[2].as_ref().unwrap().data, "b");
}

#[test]
fn accessors_reflect_parser_state() {
    let script: Script = vec![
        Step::Chunk(b"retry: 250\nid: 5\n\ndata: x\n\n".to_vec()),
        Step::End,
    ];
    let mut stream = EventStream::new(ScriptedStream::new(script));
    stream.set_last_event_id("seed");
    assert_eq!(stream.last_event_id(), "seed");
    let items = events(collect_stream(&mut stream));
    assert_eq!(items.len(), 1);
    let ev = items[0].as_ref().unwrap();
    assert_eq!(ev.id, "5");
    assert_eq!(ev.retry, Some(Duration::from_millis(250)));
    assert_eq!(stream.last_event_id(), "5");
    assert_eq!(stream.retry(), Some(Duration::from_millis(250)));
}

#[test]
fn seeded_id_is_reported_until_the_first_id_field() {
    let script: Script = vec![Step::Chunk(b"data: a\n\nid: 2\ndata: b\n\n".to_vec())];
    let mut stream = EventStream::new(ScriptedStream::new(script));
    stream.set_last_event_id("seed");
    let items = events(collect_stream(&mut stream));
    let ids: Vec<_> = items
        .iter()
        .map(|r| r.as_ref().unwrap().id.clone())
        .collect();
    assert_eq!(ids, ["seed", "2"]);
}

#[test]
fn accepts_any_as_ref_u8_chunk_type() {
    let script: Vec<Step<&'static [u8], std::io::Error>> =
        vec![Step::Chunk(b"data: borrowed\n\n"), Step::End];
    let items = collect_stream(EventStream::new(ScriptedStream::new(script)));
    assert_eq!(items.len(), 1);
    assert_eq!(items[0].as_ref().unwrap().data, "borrowed");

    let script: Vec<Step<String, String>> = vec![Step::Chunk("data: owned\n\n".to_string())];
    let items = collect_stream(EventStream::new(ScriptedStream::new(script)));
    assert_eq!(items[0].as_ref().unwrap().data, "owned");
}

#[test]
fn error_type_is_transport_only_and_displays_the_inner_error() {
    let err: Error<std::io::Error> =
        Error::Transport(std::io::Error::new(std::io::ErrorKind::Other, "wire"));
    assert_eq!(err.to_string(), "transport error: wire");
    let source = std::error::Error::source(&err).expect("source");
    assert_eq!(source.to_string(), "wire");
    let Error::Transport(inner) = err;
    assert_eq!(inner.kind(), std::io::ErrorKind::Other);
}

#[test]
fn empty_inner_stream_yields_nothing() {
    let script: Script = vec![Step::End];
    let items = collect_stream(EventStream::new(ScriptedStream::new(script)));
    assert!(items.is_empty());
}

#[test]
fn event_stream_is_send_and_unpin_when_inner_is() {
    fn assert_send_unpin<T: Send + Unpin>(_: &T) {}
    let script: Script = vec![];
    let stream = EventStream::new(ScriptedStream::new(script));
    assert_send_unpin(&stream);
}