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
111
112
113

//! Performance gate, the D1 regression: sub-quadratic parsing under
//! 1-byte chunking. The deterministic proof is the `bytes_scanned` counter, bounded
//! by 2 * input + O(1). The wall-clock growth ratio is the backstop.

mod common;

use std::time::{Duration, Instant};

use common::load_vectors;
use sse_parse::SseParser;

fn bytes_scanned_for(chunks: &[Vec<u8>]) -> (u64, usize) {
    let mut parser = SseParser::new();
    let mut total = 0usize;
    let mut events = 0usize;
    for chunk in chunks {
        parser.feed(chunk);
        total += chunk.len();
        while parser.next_event().is_some() {
            events += 1;
        }
    }
    parser.finish();
    while parser.next_event().is_some() {
        events += 1;
    }
    let _ = events;
    (parser.stats().bytes_scanned, total)
}

#[test]
fn d1_g02_huge_line_one_byte_chunks_scans_each_byte_at_most_twice() {
    let vectors = load_vectors();
    let g02 = vectors
        .iter()
        .find(|v| v.id.starts_with("g02"))
        .expect("g02 present");
    assert!(g02.chunks.iter().all(|c| c.len() == 1));
    let (scanned, total) = bytes_scanned_for(&g02.chunks);
    assert_eq!(total, 6 + 65536 + 2);
    assert!(
        scanned <= 2 * total as u64 + 64,
        "g02: scanned {scanned} bytes for {total} input bytes"
    );
}

#[test]
fn d1_g01_many_events_one_byte_chunks_scans_each_byte_at_most_twice() {
    let vectors = load_vectors();
    let g01 = vectors
        .iter()
        .find(|v| v.id.starts_with("g01"))
        .expect("g01 present");
    let (scanned, total) = bytes_scanned_for(&g01.chunks);
    assert_eq!(total, 20 * 2000);
    assert!(
        scanned <= 2 * total as u64 + 64,
        "g01: scanned {scanned} bytes for {total} input bytes"
    );
}

#[test]
fn d1_scan_counter_is_linear_for_every_vector() {
    for v in load_vectors() {
        let (scanned, total) = bytes_scanned_for(&v.chunks);
        assert!(
            scanned <= 2 * total as u64 + 64,
            "{}: scanned {scanned} for {total}",
            v.id
        );
    }
}

fn long_line_stream(fill: usize) -> Vec<u8> {
    let mut s = Vec::with_capacity(fill + 16);
    s.extend_from_slice(b"data: ");
    s.resize(s.len() + fill, b'a');
    s.extend_from_slice(b"\n\n");
    s
}

/// Minimum wall-clock time over several runs of feeding `bytes` one byte at
/// a time and draining.
fn min_time_one_byte_chunks(bytes: &[u8], runs: usize) -> Duration {
    let mut best = Duration::MAX;
    for _ in 0..runs {
        let start = Instant::now();
        let mut parser = SseParser::new();
        for b in bytes {
            parser.feed(std::slice::from_ref(b));
        }
        parser.finish();
        let ev = parser.next_event().expect("one event");
        assert_eq!(ev.data.len(), bytes.len() - 8);
        best = best.min(start.elapsed());
    }
    best
}

#[test]
fn d1_wall_clock_growth_is_sub_quadratic() {
    let small = long_line_stream(16 * 1024);
    let large = long_line_stream(64 * 1024);
    // Warm up allocator and caches.
    let _ = min_time_one_byte_chunks(&small, 1);
    let t_small = min_time_one_byte_chunks(&small, 7);
    let t_large = min_time_one_byte_chunks(&large, 7);
    let ratio = t_large.as_secs_f64() / t_small.as_secs_f64().max(1e-9);
    assert!(
        ratio < 8.0,
        "T(64 KiB) / T(16 KiB) = {ratio:.2} ({t_large:?} / {t_small:?}); quadratic would be about 16"
    );
}