sse-parse 0.1.0

//! The `max_buffer` knob: the default is unlimited, a cap above every
//! conformance vector leaves every outcome unchanged, and an exceeded cap
//! drops the block and resynchronizes at the next blank line.

mod common;

use common::{load_vectors, run_parser, run_parser_with, split_by_sizes};
use proptest::prelude::*;
use sse_parse::{Event, SseParser};

fn msg(data: &str, id: &str) -> Event {
    Event {
        event: "message".into(),
        data: data.into(),
        id: id.into(),
        retry: None,
    }
}

#[test]
fn a_cap_above_every_vector_leaves_every_outcome_unchanged() {
    for v in load_vectors() {
        let plain = run_parser(&v.chunks);
        for cap in [1 << 20, usize::MAX] {
            let capped = run_parser_with(SseParser::new().max_buffer(cap), &v.chunks);
            assert_eq!(capped, plain, "{}: cap {cap}", v.id);
        }
        let capped = SseParser::new().max_buffer(1 << 20);
        assert_eq!(capped.overflows(), 0);
    }
}

#[test]
fn oversized_block_is_dropped_and_parsing_resumes_at_next_blank_line() {
    let mut input = b"data: ok\n\ndata: ".to_vec();
    input.resize(input.len() + 200, b'x');
    input.extend_from_slice(b"\ndata: same block\n\ndata: after\n\n");
    for sizes in [&[usize::MAX][..], &[1], &[7], &[64]] {
        let mut parser = SseParser::new().max_buffer(64);
        let got = run_parser_with(parser.clone(), &split_by_sizes(&input, sizes));
        assert_eq!(
            got.events,
            vec![msg("ok", ""), msg("after", "")],
            "sizes {sizes:?}"
        );
        parser.feed(&input);
        assert_eq!(parser.overflows(), 1, "sizes {sizes:?}");
    }
}

#[test]
fn dropped_line_terminator_is_not_mistaken_for_a_blank_line() {
    // The oversized line ends with "\n" that arrives alone in its own chunk,
    // right after the pending bytes were dropped.
    let mut parser = SseParser::new().max_buffer(16);
    parser.feed(b"data: ");
    parser.feed(&[b'x'; 40]);
    assert_eq!(parser.overflows(), 1);
    parser.feed(b"\n");
    parser.feed(b"data: still dropped\n");
    parser.feed(b"\n");
    parser.feed(b"data: ok\n\n");
    let mut events = Vec::new();
    while let Some(ev) = parser.next_event() {
        events.push(ev);
    }
    assert_eq!(events, vec![msg("ok", "")]);
}

#[test]
fn many_small_data_lines_count_toward_the_cap() {
    let mut parser = SseParser::new().max_buffer(32);
    for _ in 0..10 {
        parser.feed(b"data: 0123456789\n");
    }
    parser.feed(b"\ndata: ok\n\n");
    assert_eq!(parser.overflows(), 1);
    let mut events = Vec::new();
    while let Some(ev) = parser.next_event() {
        events.push(ev);
    }
    assert_eq!(events, vec![msg("ok", "")]);
}

#[test]
fn fields_processed_before_overflow_keep_their_effect() {
    let mut parser = SseParser::new().max_buffer(32);
    parser.feed(b"id: 77\nretry: 9\n");
    parser.feed(&[b'y'; 64]);
    parser.feed(b"\nid: 88\n\ndata: next\n\n");
    assert_eq!(parser.overflows(), 1);
    let ev = parser.next_event().unwrap();
    assert_eq!(ev.id, "77");
    assert_eq!(ev.retry, Some(std::time::Duration::from_millis(9)));
    assert_eq!(parser.last_event_id(), "77");
}

#[test]
fn cap_exactly_at_line_size_does_not_trigger() {
    // "data: abcdef" is 12 raw bytes on an empty block.
    let mut parser = SseParser::new().max_buffer(12);
    parser.feed(b"data: abcdef\n\n");
    assert_eq!(parser.overflows(), 0);
    assert_eq!(parser.next_event().unwrap().data, "abcdef");

    let mut parser = SseParser::new().max_buffer(11);
    parser.feed(b"data: abcdef\n\n");
    assert_eq!(parser.overflows(), 1);
    assert!(parser.next_event().is_none());
}

#[test]
fn cap_counts_raw_line_bytes_plus_block_buffers() {
    // After "data: abcdef" the data buffer holds "abcdef\n", 7 bytes. The
    // next line "data: xy" is 8 raw bytes: 15 on top of 7 exceeds 14.
    let mut parser = SseParser::new().max_buffer(14);
    parser.feed(b"data: abcdef\ndata: xy\n\n");
    assert_eq!(parser.overflows(), 1);
    assert!(parser.next_event().is_none());

    let mut parser = SseParser::new().max_buffer(15);
    parser.feed(b"data: abcdef\ndata: xy\n\n");
    assert_eq!(parser.overflows(), 0);
    assert_eq!(parser.next_event().unwrap().data, "abcdef\nxy");
}

#[test]
fn cap_applies_when_the_whole_block_arrives_in_one_chunk() {
    // A block over the cap is dropped even when its bytes never sit in the
    // input buffer as an unterminated line.
    let mut input = b"data: ok\n\ndata: ".to_vec();
    input.resize(input.len() + 200, b'x');
    input.extend_from_slice(b"\n\ndata: after\n\n");
    let got = run_parser_with(SseParser::new().max_buffer(64), &[input]);
    assert_eq!(got.events, vec![msg("ok", ""), msg("after", "")]);
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(1500))]

    #[test]
    fn capped_parser_outcome_is_chunking_invariant(
        bytes in prop::collection::vec(sse_byte(), 0..300),
        sizes in prop::collection::vec(1usize..24, 1..32),
        cap in 0usize..64,
    ) {
        let whole = run_parser_with(SseParser::new().max_buffer(cap), std::slice::from_ref(&bytes));
        let split = run_parser_with(
            SseParser::new().max_buffer(cap),
            &split_by_sizes(&bytes, &sizes),
        );
        prop_assert_eq!(split, whole);
    }
}

/// Bytes weighted toward the SSE grammar so lines and blank lines are common.
fn sse_byte() -> impl Strategy<Value = u8> {
    prop_oneof![
        8 => Just(b'\n'),
        4 => Just(b'\r'),
        4 => Just(b':'),
        3 => Just(b' '),
        6 => prop::sample::select(&b"dataeventidretry0123456789"[..]),
        3 => any::<u8>(),
    ]
}

#[test]
fn held_bytes_never_exceed_cap_plus_one_chunk() {
    let mut parser = SseParser::new().max_buffer(100);
    for _ in 0..1000 {
        parser.feed(&[b'z'; 50]);
        assert!(parser.stats().buffered_bytes <= 150);
    }
    parser.feed(b"\n\ndata: tail\n\n");
    assert_eq!(parser.next_event().unwrap().data, "tail");
    assert_eq!(parser.overflows(), 1);
}

#[test]
fn decoded_expansion_past_the_cap_drops_the_block() {
    // "data:" plus 20 invalid bytes is 25 raw bytes, under a cap of 32, and
    // decodes to 60 bytes of U+FFFD plus the LF, over it.
    let mut input = b"data:".to_vec();
    input.resize(input.len() + 20, 0xFF);
    input.extend_from_slice(b"\n\ndata: after\n\n");
    for sizes in [&[usize::MAX][..], &[1], &[9]] {
        let got = run_parser_with(
            SseParser::new().max_buffer(32),
            &split_by_sizes(&input, sizes),
        );
        assert_eq!(got.events, vec![msg("after", "")], "sizes {sizes:?}");
    }
    let mut parser = SseParser::new().max_buffer(32);
    for chunk in split_by_sizes(&input, &[1]) {
        parser.feed(&chunk);
        assert!(parser.stats().buffered_bytes <= 33);
    }
    assert_eq!(parser.overflows(), 1);
}