qubit-http 0.9.0

General-purpose HTTP infrastructure for Rust with unified client semantics, secure logging, and built-in SSE decoding
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
115
116

/*******************************************************************************
 *
 *    Copyright (c) 2025 - 2026 Haixing Hu.
 *
 *    SPDX-License-Identifier: Apache-2.0
 *
 *    Licensed under the Apache License, Version 2.0.
 *
 ******************************************************************************/

use bytes::Bytes;
use futures_util::StreamExt;
use http::{
    HeaderMap,
    Method,
    StatusCode,
};
use qubit_http::{
    HttpErrorKind,
    HttpResponse,
};

fn stream_response_from_chunks(chunks: Vec<String>) -> HttpResponse {
    let body = chunks.join("");
    HttpResponse::new(
        StatusCode::OK,
        HeaderMap::new(),
        Bytes::from(body),
        url::Url::parse("https://example.com/stream").unwrap(),
        Method::GET,
    )
}

#[tokio::test]
async fn test_decode_events_accepts_cr_only_line_endings() {
    let response = stream_response_from_chunks(vec!["data: one\r\rdata: two\r\r".to_string()]);
    let mut events = response.sse_max_line_bytes(64).sse_max_frame_bytes(1024).sse_messages();

    let first = events.next().await.unwrap().unwrap();
    let second = events.next().await.unwrap().unwrap();
    assert_eq!(first.data, "one");
    assert_eq!(second.data, "two");
    assert!(events.next().await.is_none());
}

#[tokio::test]
async fn test_decode_events_accepts_crlf_split_across_chunks() {
    let response = stream_response_from_chunks(vec![
        "data: one\r".to_string(),
        "\r".to_string(),
        "\ndata: two\r".to_string(),
        "\n\r".to_string(),
        "\n".to_string(),
    ]);
    let mut events = response.sse_max_line_bytes(64).sse_max_frame_bytes(1024).sse_messages();

    let first = events.next().await.unwrap().unwrap();
    let second = events.next().await.unwrap().unwrap();
    assert_eq!(first.data, "one");
    assert_eq!(second.data, "two");
    assert!(events.next().await.is_none());
}

#[tokio::test]
async fn test_decode_events_accepts_dense_mixed_line_endings_in_one_chunk() {
    let response =
        stream_response_from_chunks(vec!["event: add\ndata: one\n\revent: add\rdata: two\r\r\n".to_string()]);
    let mut events = response.sse_max_line_bytes(64).sse_max_frame_bytes(1024).sse_messages();

    let first = events.next().await.unwrap().unwrap();
    let second = events.next().await.unwrap().unwrap();

    assert_eq!(first.event.as_deref(), Some("add"));
    assert_eq!(first.data, "one");
    assert_eq!(second.event.as_deref(), Some("add"));
    assert_eq!(second.data, "two");
    assert!(events.next().await.is_none());
}

#[tokio::test]
async fn test_decode_events_with_limits_rejects_line_exceeding_max_bytes() {
    let long_line = format!("data: {}\n\n", "a".repeat(64));
    let response = stream_response_from_chunks(vec![long_line]);
    let mut events = response.sse_max_line_bytes(16).sse_max_frame_bytes(1024).sse_messages();

    let error = events.next().await.unwrap().unwrap_err();
    assert_eq!(error.kind, HttpErrorKind::SseProtocol);
    assert!(error.message.contains("max_line_bytes"));
}

#[tokio::test]
async fn test_decode_events_rejects_invalid_utf8_line() {
    let response = HttpResponse::new(
        StatusCode::OK,
        HeaderMap::new(),
        Bytes::from_static(b"data: \xFF\n\n"),
        url::Url::parse("https://example.com/stream").expect("valid URL"),
        Method::GET,
    );
    let mut events = response.sse_max_line_bytes(64).sse_max_frame_bytes(1024).sse_messages();

    let error = events.next().await.unwrap().unwrap_err();

    assert_eq!(error.kind, HttpErrorKind::SseProtocol);
    assert!(error.message.contains("UTF-8"));
}

#[tokio::test]
async fn test_decode_events_with_limits_accepts_line_within_max_bytes() {
    let response = stream_response_from_chunks(vec!["data: ok\n\n".to_string()]);
    let mut events = response.sse_max_line_bytes(64).sse_max_frame_bytes(1024).sse_messages();

    let event = events.next().await.unwrap().unwrap();
    assert_eq!(event.data, "ok");
    assert!(events.next().await.is_none());
}