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
105
106
107
108
109
110

//! Chunking-invariance gate: the event sequence and final state never
//! depend on where chunk boundaries fall.

mod common;

use common::{load_vectors, run_parser, run_parser_whole, split_at_points, split_by_sizes};
use proptest::prelude::*;

#[test]
fn every_two_split_of_short_vectors_matches_single_chunk() {
    for v in load_vectors().iter().filter(|v| !v.is_generator) {
        let bytes = v.stream_bytes();
        if bytes.len() >= 64 {
            continue;
        }
        let whole = run_parser_whole(&bytes);
        assert_eq!(whole.events, v.events, "{}", v.id);
        for cut in 0..=bytes.len() {
            let got = run_parser(&split_at_points(&bytes, &[cut]));
            assert_eq!(got, whole, "{}: split at {cut}", v.id);
        }
    }
}

#[test]
fn every_three_split_of_short_vectors_matches_single_chunk() {
    for v in load_vectors().iter().filter(|v| !v.is_generator) {
        let bytes = v.stream_bytes();
        if bytes.len() >= 40 {
            continue;
        }
        let whole = run_parser_whole(&bytes);
        for a in 0..=bytes.len() {
            for b in a..=bytes.len() {
                let got = run_parser(&split_at_points(&bytes, &[a, b]));
                assert_eq!(got, whole, "{}: split at {a},{b}", v.id);
            }
        }
    }
}

#[test]
fn generator_vectors_match_under_other_chunkings() {
    for v in load_vectors().iter().filter(|v| v.is_generator) {
        let bytes = v.stream_bytes();
        let whole = run_parser_whole(&bytes);
        assert_eq!(whole.events, v.events, "{}", v.id);
        for sizes in [&[1usize][..], &[7, 3, 1], &[4096], &[13, 1, 1, 1000]] {
            let got = run_parser(&split_by_sizes(&bytes, sizes));
            assert_eq!(got, whole, "{}: sizes {sizes:?}", v.id);
        }
    }
}

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

    #[test]
    fn arbitrary_bytes_are_chunking_invariant(
        bytes in prop::collection::vec(any::<u8>(), 0..256),
        sizes in prop::collection::vec(1usize..24, 1..32),
    ) {
        let whole = run_parser_whole(&bytes);
        let got = run_parser(&split_by_sizes(&bytes, &sizes));
        prop_assert_eq!(got, whole);
    }

    #[test]
    fn sse_shaped_bytes_are_chunking_invariant(
        bytes in prop::collection::vec(sse_byte(), 0..300),
        sizes in prop::collection::vec(1usize..12, 1..40),
    ) {
        let whole = run_parser_whole(&bytes);
        let got = run_parser(&split_by_sizes(&bytes, &sizes));
        prop_assert_eq!(got, whole);
    }

    #[test]
    fn conformance_streams_are_chunking_invariant(
        index in 0usize..36,
        sizes in prop::collection::vec(1usize..40, 1..40),
    ) {
        let vectors = load_vectors();
        let v = &vectors[index];
        let bytes = v.stream_bytes();
        let whole = run_parser_whole(&bytes);
        prop_assert_eq!(&whole.events, &v.events);
        let got = run_parser(&split_by_sizes(&bytes, &sizes));
        prop_assert_eq!(got, whole);
    }
}

/// Bytes weighted toward the SSE grammar: terminators, colons, spaces,
/// field names, BOM bytes, NUL, and invalid UTF-8.
fn sse_byte() -> impl Strategy<Value = u8> {
    prop_oneof![
        8 => Just(b'\n'),
        4 => Just(b'\r'),
        4 => Just(b':'),
        3 => Just(b' '),
        2 => Just(0xEF),
        1 => Just(0xBB),
        1 => Just(0xBF),
        1 => Just(0x00),
        1 => Just(0xFF),
        1 => Just(0xC3),
        6 => prop::sample::select(&b"dataeventidretry0123456789"[..]),
        3 => any::<u8>(),
    ]
}