nemo-relay 0.6.0

Core Rust SDK for NeMo Relay observability, scope management, and runtime instrumentation.
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
114

// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Unit tests for streaming in the NeMo Relay core crate.

use super::*;
use serde_json::json;

#[test]
fn decodes_complete_frames_in_one_push() {
    let mut decoder = SseEventDecoder::new();
    let events = decoder
        .push_bytes(
            b"event: ping\ndata: {\"type\":\"ping\"}\n\nevent: msg\ndata: {\"text\":\"hi\"}\n\n",
        )
        .unwrap();
    assert_eq!(events.len(), 2);
    assert_eq!(events[0].event.as_deref(), Some("ping"));
    assert_eq!(events[0].data, json!({"type": "ping"}));
    assert_eq!(events[1].event.as_deref(), Some("msg"));
    assert_eq!(events[1].data, json!({"text": "hi"}));
}

#[test]
fn buffers_partial_frames_across_pushes() {
    let mut decoder = SseEventDecoder::new();
    assert!(decoder.push_bytes(b"event: m\ndata: ").unwrap().is_empty());
    assert!(decoder.push_bytes(b"{\"a\":1").unwrap().is_empty());
    let events = decoder.push_bytes(b"}\n\n").unwrap();
    assert_eq!(events.len(), 1);
    assert_eq!(events[0].data, json!({"a": 1}));
}

#[test]
fn normalizes_crlf_terminator_split_across_pushes() {
    let mut decoder = SseEventDecoder::new();
    assert!(
        decoder
            .push_bytes_results(b"data: {\"a\":1}\r\n\r")
            .is_empty()
    );

    let results = decoder.push_bytes_results(b"\n");
    assert_eq!(results.len(), 1);
    assert_eq!(
        results.into_iter().next().unwrap().unwrap().data,
        json!({"a": 1})
    );
}

#[test]
fn drops_frames_without_data_lines() {
    let mut decoder = SseEventDecoder::new();
    // A heartbeat-style comment frame plus a real one.
    let events = decoder
        .push_bytes(b": keepalive\n\nevent: real\ndata: {\"v\":2}\n\n")
        .unwrap();
    assert_eq!(events.len(), 1);
    assert_eq!(events[0].data, json!({"v": 2}));
}

#[test]
fn surfaces_final_partial_frame_on_finish() {
    let mut decoder = SseEventDecoder::new();
    decoder
        .push_bytes(b"event: tail\ndata: {\"end\":true}")
        .unwrap();
    let trailing = decoder.finish().unwrap().expect("trailing frame present");
    assert_eq!(trailing.data, json!({"end": true}));
}

#[test]
fn drops_openai_chat_done_sentinel() {
    let mut decoder = SseEventDecoder::new();
    let events = decoder
        .push_bytes(
            b"data: {\"id\":\"chatcmpl-1\"}\n\ndata: [DONE]\n\ndata: {\"id\":\"chatcmpl-2\"}\n\n",
        )
        .unwrap();
    // [DONE] is dropped; surrounding JSON events still come through.
    assert_eq!(events.len(), 2);
    assert_eq!(events[0].data, json!({"id": "chatcmpl-1"}));
    assert_eq!(events[1].data, json!({"id": "chatcmpl-2"}));
}

#[test]
fn surfaces_parse_errors_with_payload_context() {
    let mut decoder = SseEventDecoder::new();
    let error = decoder
        .push_bytes(b"event: bad\ndata: {not valid json}\n\n")
        .unwrap_err();
    let message = error.to_string();
    assert!(message.contains("SSE data payload"), "{message}");
    assert!(message.contains("not valid json"), "{message}");
}

#[test]
fn preserves_successes_before_a_later_parse_error() {
    let mut decoder = SseEventDecoder::new();
    let mut results = decoder
        .push_bytes_results(
            b"event: good\ndata: {\"chunk\":\"first\"}\n\nevent: bad\ndata: {not valid json}\n\n",
        )
        .into_iter();

    let first = results.next().unwrap().unwrap();
    assert_eq!(first.event.as_deref(), Some("good"));
    assert_eq!(first.data, json!({"chunk": "first"}));

    let error = results.next().unwrap().unwrap_err().to_string();
    assert!(error.contains("SSE data payload"), "{error}");
    assert!(error.contains("not valid json"), "{error}");
    assert!(results.next().is_none());
}