sse-parse 0.1.0

Incremental, linear-time parser for Server-Sent Events byte streams, with a sans-IO core and a runtime-agnostic Stream adapter
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

//! Totality gate: `feed`, `finish`, and `next_event` are total over
//! arbitrary bytes and chunkings, and memory stays proportional to the input.

mod common;

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

fn feed_all(parser: &mut SseParser, chunks: &[Vec<u8>], total: usize) -> usize {
    let mut events = 0;
    let mut fed = 0;
    for chunk in chunks {
        parser.feed(chunk);
        fed += chunk.len();
        // Each invalid byte decodes to the three-byte U+FFFD, so the block
        // buffers can hold up to three bytes per byte fed.
        let held = parser.stats().buffered_bytes;
        assert!(
            held <= 3 * fed + 8,
            "holding {held} bytes after feeding {fed} of {total}"
        );
        while let Some(ev) = parser.next_event() {
            assert!(!ev.event.is_empty());
            events += 1;
        }
    }
    parser.finish();
    assert_eq!(parser.stats().buffered_bytes, 0);
    while parser.next_event().is_some() {
        events += 1;
    }
    events
}

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

    #[test]
    fn arbitrary_input_never_panics_and_memory_is_bounded(
        bytes in prop::collection::vec(any::<u8>(), 0..512),
        sizes in prop::collection::vec(1usize..64, 1..16),
    ) {
        let chunks = split_by_sizes(&bytes, &sizes);
        let mut parser = SseParser::new();
        let events = feed_all(&mut parser, &chunks, bytes.len());
        prop_assert_eq!(events, run_parser(&chunks).events.len());
    }

    #[test]
    fn terminator_heavy_input_never_panics(
        bytes in prop::collection::vec(
            prop_oneof![Just(b'\r'), Just(b'\n'), Just(b':'), Just(b' '), Just(0xEF), any::<u8>()],
            0..512,
        ),
        sizes in prop::collection::vec(1usize..8, 1..16),
    ) {
        let chunks = split_by_sizes(&bytes, &sizes);
        let mut parser = SseParser::new();
        feed_all(&mut parser, &chunks, bytes.len());
    }

    #[test]
    fn capped_parser_never_panics_and_holds_at_most_cap_plus_chunk(
        bytes in prop::collection::vec(any::<u8>(), 0..512),
        sizes in prop::collection::vec(1usize..64, 1..16),
        cap in 0usize..128,
    ) {
        let chunks = split_by_sizes(&bytes, &sizes);
        let mut parser = SseParser::new().max_buffer(cap);
        let max_chunk = sizes.iter().copied().max().unwrap_or(1);
        for chunk in &chunks {
            parser.feed(chunk);
            let held = parser.stats().buffered_bytes;
            prop_assert!(held <= cap + max_chunk + 1, "held {held} with cap {cap}");
            while parser.next_event().is_some() {}
        }
        parser.finish();
        while parser.next_event().is_some() {}
    }
}

#[test]
fn seeded_id_never_panics_on_junk() {
    let mut parser = SseParser::with_last_event_id("seed");
    parser.feed(&[0xFF, 0xFE, 0x00, b'\r', b'\r', b'\n', 0xEF, 0xBB]);
    parser.finish();
    parser.finish();
    while parser.next_event().is_some() {}
    assert_eq!(parser.last_event_id(), "seed");
}

#[test]
fn empty_feeds_are_harmless() {
    let mut parser = SseParser::new();
    parser.feed(b"");
    parser.feed(b"data: x");
    parser.feed(b"");
    parser.feed(b"\n\n");
    parser.feed(b"");
    assert_eq!(parser.next_event().unwrap().data, "x");
    parser.finish();
    assert!(parser.next_event().is_none());
}