sse-parse 0.1.0

//! Shared helpers for the integration tests: conformance vector loading,
//! parser and stream runners, and a dependency-free executor.

#![allow(dead_code)]

use std::collections::VecDeque;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use std::time::Duration;

use futures_core::Stream;
use serde_json::Value;
use sse_parse::{Error, Event, EventStream, SseParser};

pub const VECTORS_JSON: &str = include_str!("../conformance-vectors.json");

/// One conformance vector, decoded into bytes and expected outputs.
#[derive(Debug, Clone)]
pub struct Vector {
    pub id: String,
    pub purpose: String,
    /// Declared chunking: one stream item per entry.
    pub chunks: Vec<Vec<u8>>,
    pub events: Vec<Event>,
    pub final_last_event_id: String,
    pub retry_updates: Vec<Duration>,
    pub is_generator: bool,
}

impl Vector {
    pub fn stream_bytes(&self) -> Vec<u8> {
        self.chunks.concat()
    }
}

/// Observable outcome of one parse, compared across chunkings.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outcome {
    pub events: Vec<Event>,
    pub last_event_id: String,
    pub retry: Option<Duration>,
    pub retry_updates: u64,
}

pub fn load_vectors() -> Vec<Vector> {
    let root: Value = serde_json::from_str(VECTORS_JSON).expect("conformance-vectors.json parses");
    // The schema tag is part of the frozen fixture bytes; match it verbatim.
    assert_eq!(root["schema"], "sse-conformance-vectors/1");
    let vectors = root["vectors"].as_array().expect("vectors array");
    let out: Vec<Vector> = vectors.iter().map(decode_vector).collect();
    assert_eq!(out.len(), 36, "34 explicit vectors + 2 generators");
    out
}

fn decode_vector(v: &Value) -> Vector {
    let id = v["id"].as_str().expect("id").to_string();
    let purpose = v["purpose"].as_str().unwrap_or("").to_string();
    let expected = &v["expected"];
    let (chunks, is_generator) = if let Some(chunks) = v["chunks"].as_array() {
        (chunks.iter().map(decode_chunk).collect(), false)
    } else {
        (decode_generator(&v["generator"]), true)
    };
    let events = decode_expected_events(&expected["events"]);
    let final_last_event_id = expected["final_last_event_id"]
        .as_str()
        .expect("final_last_event_id")
        .to_string();
    let retry_updates = expected["retry_updates"]
        .as_array()
        .expect("retry_updates")
        .iter()
        .map(|n| Duration::from_millis(n.as_u64().expect("retry ms")))
        .collect();
    Vector {
        id,
        purpose,
        chunks,
        events,
        final_last_event_id,
        retry_updates,
        is_generator,
    }
}

fn decode_chunk(c: &Value) -> Vec<u8> {
    if let Some(text) = c["text"].as_str() {
        return text.as_bytes().to_vec();
    }
    let hex = c["hex"].as_str().expect("chunk is text or hex");
    assert!(hex.len() % 2 == 0, "even hex length");
    (0..hex.len())
        .step_by(2)
        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("hex digit pair"))
        .collect()
}

fn chunked(bytes: &[u8], chunk_bytes: usize) -> Vec<Vec<u8>> {
    bytes.chunks(chunk_bytes).map(<[u8]>::to_vec).collect()
}

fn decode_generator(g: &Value) -> Vec<Vec<u8>> {
    let chunk_bytes = g["chunk_bytes"].as_u64().expect("chunk_bytes") as usize;
    match g["kind"].as_str().expect("generator kind") {
        "repeat_stream" => {
            let unit = g["unit_text"].as_str().expect("unit_text");
            let count = g["count"].as_u64().expect("count") as usize;
            chunked(unit.repeat(count).as_bytes(), chunk_bytes)
        }
        "long_line" => {
            let mut s = g["prefix"].as_str().expect("prefix").to_string();
            let fill = g["fill_char"].as_str().expect("fill_char");
            let n = g["fill_count"].as_u64().expect("fill_count") as usize;
            s.push_str(&fill.repeat(n));
            s.push_str(g["suffix"].as_str().expect("suffix"));
            chunked(s.as_bytes(), chunk_bytes)
        }
        other => panic!("unknown generator kind {other}"),
    }
}

fn decode_event(e: &Value) -> Event {
    Event {
        event: e["event"].as_str().expect("event").to_string(),
        data: e["data"].as_str().expect("data").to_string(),
        id: e["id"].as_str().expect("id").to_string(),
        retry: e["retry"].as_u64().map(Duration::from_millis),
    }
}

fn decode_expected_events(e: &Value) -> Vec<Event> {
    if let Some(list) = e.as_array() {
        return list.iter().map(decode_event).collect();
    }
    match e["kind"].as_str().expect("events kind") {
        "repeat" => {
            let ev = decode_event(&e["event"]);
            let count = e["count"].as_u64().expect("count") as usize;
            vec![ev; count]
        }
        "single_fill_event" => {
            let fill = e["fill_char"].as_str().expect("fill_char");
            let n = e["fill_count"].as_u64().expect("fill_count") as usize;
            vec![Event {
                event: e["event"].as_str().expect("event").to_string(),
                data: fill.repeat(n),
                id: e["id"].as_str().expect("id").to_string(),
                retry: e["retry"].as_u64().map(Duration::from_millis),
            }]
        }
        other => panic!("unknown events kind {other}"),
    }
}

/// Feed the chunks into `parser`, finish, and collect everything observable.
pub fn run_parser_with(mut parser: SseParser, chunks: &[Vec<u8>]) -> Outcome {
    let mut events = Vec::new();
    for chunk in chunks {
        parser.feed(chunk);
        while let Some(ev) = parser.next_event() {
            events.push(ev);
        }
    }
    parser.finish();
    while let Some(ev) = parser.next_event() {
        events.push(ev);
    }
    Outcome {
        events,
        last_event_id: parser.last_event_id().to_string(),
        retry: parser.retry(),
        retry_updates: parser.stats().retry_updates,
    }
}

pub fn run_parser(chunks: &[Vec<u8>]) -> Outcome {
    run_parser_with(SseParser::new(), chunks)
}

pub fn run_parser_whole(bytes: &[u8]) -> Outcome {
    run_parser(&[bytes.to_vec()])
}

/// Feed one byte at a time and record every reconnection-time update in
/// order. At most one line completes per byte, so at most one update can
/// happen per feed, which makes the observation exact.
pub fn observe_retry_updates(bytes: &[u8]) -> Vec<Duration> {
    let mut parser = SseParser::new();
    let mut seen = 0;
    let mut updates = Vec::new();
    for b in bytes {
        parser.feed(std::slice::from_ref(b));
        let now = parser.stats().retry_updates;
        assert!(now <= seen + 1, "more than one retry update from one byte");
        if now > seen {
            updates.push(parser.retry().expect("update sets retry"));
            seen = now;
        }
    }
    parser.finish();
    assert_eq!(parser.stats().retry_updates, seen);
    updates
}

/// Split `bytes` at the given ascending cut points.
pub fn split_at_points(bytes: &[u8], cuts: &[usize]) -> Vec<Vec<u8>> {
    let mut out = Vec::new();
    let mut start = 0;
    for &cut in cuts {
        let cut = cut.min(bytes.len());
        if cut > start {
            out.push(bytes[start..cut].to_vec());
            start = cut;
        }
    }
    out.push(bytes[start..].to_vec());
    out
}

/// Split `bytes` into consecutive chunks of the given sizes (cycled), never
/// producing an empty chunk unless the input is empty.
pub fn split_by_sizes(bytes: &[u8], sizes: &[usize]) -> Vec<Vec<u8>> {
    if sizes.is_empty() {
        return vec![bytes.to_vec()];
    }
    let mut out = Vec::new();
    let mut pos = 0;
    let mut i = 0;
    while pos < bytes.len() {
        let size = sizes[i % sizes.len()].max(1);
        let end = (pos + size).min(bytes.len());
        out.push(bytes[pos..end].to_vec());
        pos = end;
        i += 1;
    }
    if out.is_empty() {
        out.push(Vec::new());
    }
    out
}

// ---- executor and test streams -------------------------------------------

struct NoopWake;

impl Wake for NoopWake {
    fn wake(self: Arc<Self>) {}
}

pub fn noop_waker() -> Waker {
    Waker::from(Arc::new(NoopWake))
}

/// One scripted response of a [`ScriptedStream`].
#[derive(Debug, Clone)]
pub enum Step<B, E> {
    Chunk(B),
    Fail(E),
    Pending,
    End,
}

/// A stream that replays a script of chunks, errors, pendings, and ends.
/// After the script runs out it yields `Ready(None)`; a `Pending` step wakes
/// the task immediately so a simple executor keeps polling.
pub struct ScriptedStream<B, E> {
    pub script: VecDeque<Step<B, E>>,
    pub polls: usize,
}

impl<B, E> ScriptedStream<B, E> {
    pub fn new(script: Vec<Step<B, E>>) -> Self {
        ScriptedStream {
            script: script.into(),
            polls: 0,
        }
    }
}

impl<B: Unpin, E: Unpin> Stream for ScriptedStream<B, E> {
    type Item = Result<B, E>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        this.polls += 1;
        match this.script.pop_front() {
            Some(Step::Chunk(b)) => Poll::Ready(Some(Ok(b))),
            Some(Step::Fail(e)) => Poll::Ready(Some(Err(e))),
            Some(Step::Pending) => {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
            Some(Step::End) | None => Poll::Ready(None),
        }
    }
}

pub fn chunk_steps(chunks: &[Vec<u8>]) -> Vec<Step<Vec<u8>, String>> {
    chunks.iter().cloned().map(Step::Chunk).collect()
}

/// Poll `stream` to completion on a no-op waker, collecting every item.
pub fn collect_stream<S: Stream + Unpin>(mut stream: S) -> Vec<S::Item> {
    let waker = noop_waker();
    let mut cx = Context::from_waker(&waker);
    let mut items = Vec::new();
    let mut pending_polls = 0usize;
    loop {
        match Pin::new(&mut stream).poll_next(&mut cx) {
            Poll::Ready(Some(item)) => items.push(item),
            Poll::Ready(None) => return items,
            Poll::Pending => {
                pending_polls += 1;
                assert!(pending_polls < 1_000_000, "stream never completed");
            }
        }
    }
}

/// Run the chunks through `EventStream` and collect the same outcome shape
/// the parser runner produces.
pub fn run_stream(chunks: &[Vec<u8>]) -> Outcome {
    let mut stream = EventStream::new(ScriptedStream::new(chunk_steps(chunks)));
    let items = collect_stream(&mut stream);
    let events = items
        .into_iter()
        .map(|r| match r {
            Ok(ev) => ev,
            Err(Error::Transport(e)) => panic!("unexpected transport error {e}"),
        })
        .collect();
    Outcome {
        events,
        last_event_id: stream.last_event_id().to_string(),
        retry: stream.retry(),
        retry_updates: 0,
    }
}

pub fn events_of(bytes: &[u8]) -> Vec<Event> {
    run_parser_whole(bytes).events
}