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

/*******************************************************************************
 *
 *    Copyright (c) 2025 - 2026 Haixing Hu.
 *
 *    SPDX-License-Identifier: Apache-2.0
 *
 *    Licensed under the Apache License, Version 2.0.
 *
 ******************************************************************************/
//! Decode public SSE messages from an HTTP body byte stream with size limits.

use async_stream::stream;
use futures_util::StreamExt;

use crate::HttpByteStream;

use super::{
    frame_decoder,
    line_decoder,
    SseMessageStream,
    SseRecord,
    SseRecordStream,
};

/// Parses internal SSE records from a body byte stream with explicit size limits.
///
/// # Parameters
/// - `stream`: Body byte stream whose payload is SSE text.
/// - `max_line_bytes`: Maximum allowed bytes for one SSE line.
/// - `max_frame_bytes`: Maximum allowed bytes for one SSE frame.
///
/// # Returns
/// Stream yielding internal [`SseRecord`] values or protocol/transport errors.
pub(crate) fn decode_records_from_stream_with_limits(
    stream: HttpByteStream,
    max_line_bytes: usize,
    max_frame_bytes: usize,
) -> SseRecordStream {
    let lines = line_decoder::decode_lines(stream, max_line_bytes);
    frame_decoder::decode_records(lines, max_frame_bytes)
}

/// Parses public SSE messages from a body byte stream with explicit size limits.
///
/// # Parameters
/// - `stream`: Body byte stream whose payload is SSE text.
/// - `max_line_bytes`: Maximum allowed bytes for one SSE line.
/// - `max_frame_bytes`: Maximum allowed bytes for one SSE frame.
///
/// # Returns
/// Stream yielding [`super::SseMessage`] values or protocol/transport errors.
pub(crate) fn decode_messages_from_stream_with_limits(
    stream: HttpByteStream,
    max_line_bytes: usize,
    max_frame_bytes: usize,
) -> SseMessageStream {
    let mut records = decode_records_from_stream_with_limits(stream, max_line_bytes, max_frame_bytes);
    let output = stream! {
        while let Some(item) = records.next().await {
            match item {
                Ok(SseRecord::Dispatch(message)) => yield Ok(message),
                Ok(SseRecord::Control(_)) => continue,
                Err(error) => {
                    yield Err(error);
                    return;
                }
            }
        }
    };
    Box::pin(output)
}