feat(transport): add incremental streaming body reader and custom ports
- Transport::stream_request returns a ChunkStream for true SSE streaming (head read incrementally, body drained chunk by chunk) - transparent incremental chunked-transfer decoding (ChunkedDecoder) - HttpRequest.port for non-standard base_url ports (Host header + TLS connect) - per-read timeouts on the streaming path
This commit is contained in:
DaiChaoXiong
parent
32b630e440
commit
7f00886150
|
|
@ -39,6 +39,9 @@ pub struct HttpRequest {
|
|||
/// 目标主机名。
|
||||
/// The target host.
|
||||
pub host: String,
|
||||
/// 目标端口（默认 443）。
|
||||
/// The target port (default 443).
|
||||
pub port: u16,
|
||||
/// 请求头（按插入顺序保留）。
|
||||
/// Request headers, in insertion order.
|
||||
pub headers: Vec<(String, String)>,
|
||||
|
|
@ -55,11 +58,19 @@ impl HttpRequest {
|
|||
method: Method::Post,
|
||||
path: path.to_string(),
|
||||
host: host.to_string(),
|
||||
port: 443,
|
||||
headers: Vec::new(),
|
||||
body: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// 设置目标端口（用于非标准端口）。
|
||||
/// Set the target port (for non-standard ports).
|
||||
pub fn port(mut self, port: u16) -> Self {
|
||||
self.port = port;
|
||||
self
|
||||
}
|
||||
|
||||
/// 追加一个请求头，返回自身以支持链式调用。
|
||||
/// Append a header and return self for chaining.
|
||||
pub fn header(mut self, name: &str, value: &str) -> Self {
|
||||
|
|
@ -93,7 +104,11 @@ impl HttpRequest {
|
|||
.iter()
|
||||
.any(|(k, _)| k.eq_ignore_ascii_case("host"));
|
||||
if !has_host {
|
||||
out.extend_from_slice(format!("Host: {}\r\n", self.host).as_bytes());
|
||||
if self.port == 443 {
|
||||
out.extend_from_slice(format!("Host: {}\r\n", self.host).as_bytes());
|
||||
} else {
|
||||
out.extend_from_slice(format!("Host: {}:{}\r\n", self.host, self.port).as_bytes());
|
||||
}
|
||||
}
|
||||
for (k, v) in &self.headers {
|
||||
out.extend_from_slice(format!("{}: {}\r\n", k, v).as_bytes());
|
||||
|
|
@ -223,60 +238,3 @@ fn find_subsequence(haystack: &[u8], needle: &[u8]) -> Option<usize> {
|
|||
.windows(needle.len())
|
||||
.position(|window| window == needle)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn serializes_post_request() {
|
||||
let req = HttpRequest::post("api.example.com", "/v1/messages")
|
||||
.header("Authorization", "Bearer xyz")
|
||||
.json_body(r#"{"hello":"world"}"#);
|
||||
let wire = req.serialize();
|
||||
let text = String::from_utf8(wire).unwrap();
|
||||
assert!(text.starts_with("POST /v1/messages HTTP/1.1\r\n"));
|
||||
assert!(text.contains("Host: api.example.com\r\n"));
|
||||
assert!(text.contains("Authorization: Bearer xyz\r\n"));
|
||||
assert!(text.contains("Content-Type: application/json\r\n"));
|
||||
assert!(text.contains("Content-Length: 17\r\n"));
|
||||
assert!(text.ends_with("{\"hello\":\"world\"}"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_response_head() {
|
||||
let raw =
|
||||
b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 5\r\n\r\nhello";
|
||||
let (resp, consumed) = parse_response_head(raw).unwrap();
|
||||
assert_eq!(resp.status, 200);
|
||||
assert_eq!(
|
||||
resp.headers.get("content-type").unwrap(),
|
||||
"application/json"
|
||||
);
|
||||
assert_eq!(resp.headers.get("content-length").unwrap(), "5");
|
||||
assert_eq!(consumed, raw.len() - 5);
|
||||
assert_eq!(body_length(&resp), BodyLength::Fixed(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn detects_chunked() {
|
||||
let raw = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
|
||||
let (resp, _) = parse_response_head(raw).unwrap();
|
||||
assert_eq!(body_length(&resp), BodyLength::Chunked);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_success_for_2xx() {
|
||||
let resp = HttpResponse {
|
||||
status: 204,
|
||||
headers: HashMap::new(),
|
||||
body: vec![],
|
||||
};
|
||||
assert!(resp.is_success());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_incomplete_head() {
|
||||
assert!(parse_response_head(b"HTTP/1.1 200 OK\r\n").is_err());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,4 +37,4 @@ pub use http::{HttpRequest, HttpResponse, Method};
|
|||
pub use sse::{SseEvent, SseParser};
|
||||
/// HTTP 传输实现与传输 trait。
|
||||
/// HTTP transport implementation and the transport trait.
|
||||
pub use transport::{HttpTransport, Transport};
|
||||
pub use transport::{ChunkStream, HttpTransport, Transport};
|
||||
|
|
|
|||
|
|
@ -192,105 +192,3 @@ impl SseParser {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn collect(parser: &mut SseParser) -> Vec<SseEvent> {
|
||||
let mut out = Vec::new();
|
||||
while let Some(e) = parser.next_event() {
|
||||
out.push(e);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_single_event() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: hello\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events.len(), 1);
|
||||
assert_eq!(events[0].data, "hello");
|
||||
assert_eq!(events[0].event, "message");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_event_with_type() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("event: ping\ndata: 1\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].event, "ping");
|
||||
assert_eq!(events[0].data, "1");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn joins_multiline_data() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: line1\ndata: line2\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "line1\nline2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_crlf() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: hi\r\n\r\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "hi");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ignores_comments_and_unknown_fields() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str(": this is a comment\ndata: ok\nfoo: bar\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "ok");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_id_and_retry() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("id: 42\nretry: 5000\ndata: x\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].id, "42");
|
||||
assert_eq!(events[0].retry, Some(5000));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_split_across_chunks() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("dat");
|
||||
assert!(collect(&mut p).is_empty());
|
||||
p.feed_str("a: hello\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_split_across_line_boundary() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: first\n\ndata: sec");
|
||||
assert_eq!(collect(&mut p).len(), 1);
|
||||
p.feed_str("ond\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "second");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multiple_events_in_one_chunk() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: a\n\ndata: b\n\ndata: c\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events.len(), 3);
|
||||
assert_eq!(events[2].data, "c");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_data_line() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data:\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,7 +12,37 @@
|
|||
use crate::error::{TransportError, TransportResult};
|
||||
use crate::http::{body_length, parse_response_head, BodyLength, HttpRequest, HttpResponse};
|
||||
use crate::tls;
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::time::Duration;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::time::timeout;
|
||||
|
||||
/// 异步读取下一块响应体的 future 类型。
|
||||
/// Future type for reading the next response-body chunk.
|
||||
pub type ChunkFuture<'a> =
|
||||
Pin<Box<dyn Future<Output = TransportResult<Option<Vec<u8>>>> + Send + 'a>>;
|
||||
/// 普通请求的 future 类型。
|
||||
/// Future type for a plain request.
|
||||
pub type RequestFuture<'a> =
|
||||
Pin<Box<dyn Future<Output = TransportResult<HttpResponse>> + Send + 'a>>;
|
||||
/// 流式请求的 future 类型。
|
||||
/// Future type for a streaming request.
|
||||
pub type StreamRequestFuture<'a> =
|
||||
Pin<Box<dyn Future<Output = TransportResult<Box<dyn ChunkStream + Send>>> + Send + 'a>>;
|
||||
|
||||
/// 流式响应体的读取接口：调用方逐块拉取，直到返回 `None` 表示流结束。
|
||||
/// A streaming response-body reader: callers pull chunks until `None`.
|
||||
///
|
||||
/// 该接口是异步的（每次拉取都是 future），因此底层可以边收边解析，
|
||||
/// 支撑 SSE 的真流式体验。
|
||||
/// This interface is async (each pull is a future), so the underlying socket
|
||||
/// can be parsed incrementally, enabling a true SSE streaming experience.
|
||||
pub trait ChunkStream: Send {
|
||||
/// 读取下一块响应体字节；`None` 表示流结束。
|
||||
/// Read the next chunk of response body bytes; `None` means end of stream.
|
||||
fn next_chunk<'a>(&'a mut self) -> ChunkFuture<'a>;
|
||||
}
|
||||
|
||||
/// 通过网络连接发送 HTTP 请求。
|
||||
/// Sends HTTP requests over a network connection.
|
||||
|
|
@ -23,12 +53,22 @@ pub trait Transport: Send + Sync {
|
|||
/// 实现者在 future 生命周期内借用请求；调用方须使其存活至 future 完成。
|
||||
/// Implementations borrow the request for the duration of the future;
|
||||
/// callers must keep it alive until the future resolves.
|
||||
fn request<'a>(
|
||||
fn request<'a>(&'a self, req: &'a HttpRequest) -> RequestFuture<'a>;
|
||||
|
||||
/// 发送请求并返回一个流式响应体读取器。
|
||||
/// Send a request and get a streaming response-body reader back.
|
||||
///
|
||||
/// 与 [`request`](Self::request) 不同，本方法不会把整个响应体读入内存；
|
||||
/// 调用方通过 [`ChunkStream::next_chunk`] 逐块拉取。
|
||||
/// 非 2xx 状态码仍会以 [`TransportError::Http`] 返回（此时已读取完整错误体）。
|
||||
/// Unlike [`request`](Self::request), this does not buffer the whole body;
|
||||
/// callers pull chunks via [`ChunkStream::next_chunk`]. Non-2xx statuses
|
||||
/// still surface as [`TransportError::Http`] (with the full error body read).
|
||||
fn stream_request<'a>(
|
||||
&'a self,
|
||||
req: &'a HttpRequest,
|
||||
) -> std::pin::Pin<
|
||||
Box<dyn std::future::Future<Output = TransportResult<HttpResponse>> + Send + 'a>,
|
||||
>;
|
||||
timeout: Duration,
|
||||
) -> StreamRequestFuture<'a>;
|
||||
}
|
||||
|
||||
/// 基于 tokio + rustls 的真实 HTTPS 传输实现。
|
||||
|
|
@ -45,20 +85,23 @@ impl HttpTransport {
|
|||
}
|
||||
|
||||
impl Transport for HttpTransport {
|
||||
fn request<'a>(
|
||||
fn request<'a>(&'a self, req: &'a HttpRequest) -> RequestFuture<'a> {
|
||||
Box::pin(async move { perform_request(req).await })
|
||||
}
|
||||
|
||||
fn stream_request<'a>(
|
||||
&'a self,
|
||||
req: &'a HttpRequest,
|
||||
) -> std::pin::Pin<
|
||||
Box<dyn std::future::Future<Output = TransportResult<HttpResponse>> + Send + 'a>,
|
||||
> {
|
||||
Box::pin(async move { perform_request(req).await })
|
||||
timeout: Duration,
|
||||
) -> StreamRequestFuture<'a> {
|
||||
Box::pin(async move { perform_stream_request(req, timeout).await })
|
||||
}
|
||||
}
|
||||
|
||||
/// 执行单次 HTTP 请求：建立 TLS 连接、序列化并发送请求、读取完整响应并解析。
|
||||
/// Perform a single HTTP request: open TLS, serialize & send, read the full response, parse.
|
||||
async fn perform_request(req: &HttpRequest) -> TransportResult<HttpResponse> {
|
||||
let mut stream = tls::connect(&req.host, 443).await?;
|
||||
let mut stream = tls::connect(&req.host, req.port).await?;
|
||||
|
||||
let wire = req.serialize();
|
||||
stream
|
||||
|
|
@ -131,9 +174,386 @@ async fn perform_request(req: &HttpRequest) -> TransportResult<HttpResponse> {
|
|||
Ok(resp)
|
||||
}
|
||||
|
||||
/// 执行单次流式 HTTP 请求：建立 TLS 连接、序列化并发送请求、解析响应头，
|
||||
/// 然后返回一个定位在响应体起始处的流式读取器。
|
||||
/// Perform a single streaming HTTP request: open TLS, serialize & send, parse
|
||||
/// the response head, then return a reader positioned at the body start.
|
||||
///
|
||||
/// 响应头是逐字节累积解析的（直到 `\r\n\r\n`），因此无需缓冲整个响应。
|
||||
/// The response head is accumulated and parsed incrementally (until
|
||||
/// `\r\n\r\n`), so no full-response buffering is required.
|
||||
async fn perform_stream_request(
|
||||
req: &HttpRequest,
|
||||
timeout_dur: Duration,
|
||||
) -> TransportResult<Box<dyn ChunkStream + Send>> {
|
||||
let mut stream = timeout(timeout_dur, tls::connect(&req.host, req.port))
|
||||
.await
|
||||
.map_err(|_| TransportError::Timeout)??;
|
||||
|
||||
let wire = req.serialize();
|
||||
timeout(timeout_dur, stream.write_all(&wire))
|
||||
.await
|
||||
.map_err(|_| TransportError::Timeout)??;
|
||||
timeout(timeout_dur, stream.flush())
|
||||
.await
|
||||
.map_err(|_| TransportError::Timeout)??;
|
||||
|
||||
// 累积响应头，直到出现头部终止符。
|
||||
// Accumulate the response head until the header terminator appears.
|
||||
let mut head_buf: Vec<u8> = Vec::with_capacity(4096);
|
||||
let head_len = loop {
|
||||
if let Some(end) = find_subsequence(&head_buf, b"\r\n\r\n") {
|
||||
break end + 4;
|
||||
}
|
||||
if head_buf.len() > 64 * 1024 {
|
||||
return Err(TransportError::Io("response head too large".into()));
|
||||
}
|
||||
let mut chunk = [0u8; 4096];
|
||||
let n = match timeout(timeout_dur, stream.read(&mut chunk)).await {
|
||||
Err(_) => return Err(TransportError::Timeout),
|
||||
Ok(Err(e)) => return Err(TransportError::Io(format!("read response head: {}", e))),
|
||||
Ok(Ok(0)) => {
|
||||
return Err(TransportError::Io(
|
||||
"connection closed before response head".into(),
|
||||
))
|
||||
}
|
||||
Ok(Ok(n)) => n,
|
||||
};
|
||||
head_buf.extend_from_slice(&chunk[..n]);
|
||||
};
|
||||
|
||||
let (resp, consumed) = parse_response_head(&head_buf)?;
|
||||
// 头部解析已消费的字节数之后的字节，是响应体的一部分。
|
||||
// Bytes beyond the consumed head are already part of the response body.
|
||||
let body_prefix = head_buf[consumed..head_len].to_vec();
|
||||
|
||||
if !resp.is_success() {
|
||||
// 非 2xx：读取剩余错误体后返回 Http 错误。
|
||||
// Non-2xx: drain the remaining error body, then return an Http error.
|
||||
let mut body = body_prefix;
|
||||
loop {
|
||||
let mut chunk = [0u8; 4096];
|
||||
match timeout(timeout_dur, stream.read(&mut chunk)).await {
|
||||
Err(_) => return Err(TransportError::Timeout),
|
||||
Ok(Err(e)) => {
|
||||
let benign = matches!(
|
||||
e.kind(),
|
||||
std::io::ErrorKind::UnexpectedEof | std::io::ErrorKind::ConnectionReset
|
||||
);
|
||||
if !benign || body.is_empty() {
|
||||
return Err(TransportError::Io(format!("read error body: {}", e)));
|
||||
}
|
||||
break;
|
||||
}
|
||||
Ok(Ok(0)) => break,
|
||||
Ok(Ok(n)) => body.extend_from_slice(&chunk[..n]),
|
||||
}
|
||||
}
|
||||
return Err(TransportError::Http {
|
||||
status: resp.status,
|
||||
body: String::from_utf8_lossy(&body).to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let reader = StreamingBody::new(stream, body_length(&resp), body_prefix, timeout_dur);
|
||||
Ok(Box::new(reader))
|
||||
}
|
||||
|
||||
/// 定位在响应体起始处的流式读取器。透明处理分块传输编码与固定长度。
|
||||
/// A streaming reader positioned at the response-body start. Transparently
|
||||
/// handles chunked transfer encoding and fixed lengths.
|
||||
struct StreamingBody {
|
||||
stream: crate::tls::TlsConnection,
|
||||
mode: BodyLength,
|
||||
/// 头部解析后已多读的响应体字节。
|
||||
/// Body bytes already read beyond the head.
|
||||
pending: Vec<u8>,
|
||||
/// 固定长度模式下剩余字节数。
|
||||
/// Remaining bytes in fixed-length mode.
|
||||
remaining: usize,
|
||||
/// 分块模式下使用的增量解码器。
|
||||
/// Incremental decoder used in chunked mode.
|
||||
decoder: Option<ChunkedDecoder>,
|
||||
finished: bool,
|
||||
timeout_dur: Duration,
|
||||
}
|
||||
|
||||
impl StreamingBody {
|
||||
fn new(
|
||||
stream: crate::tls::TlsConnection,
|
||||
mode: BodyLength,
|
||||
pending: Vec<u8>,
|
||||
timeout_dur: Duration,
|
||||
) -> Self {
|
||||
let remaining = match mode {
|
||||
BodyLength::Fixed(n) => n,
|
||||
_ => 0,
|
||||
};
|
||||
Self {
|
||||
stream,
|
||||
mode,
|
||||
pending,
|
||||
remaining,
|
||||
decoder: (mode == BodyLength::Chunked).then(ChunkedDecoder::new),
|
||||
finished: false,
|
||||
timeout_dur,
|
||||
}
|
||||
}
|
||||
|
||||
/// 从套接字读取一块原始字节；`None` 表示干净 EOF。
|
||||
/// Read one raw chunk from the socket; `None` is a clean EOF.
|
||||
async fn read_raw(&mut self) -> TransportResult<Option<Vec<u8>>> {
|
||||
let mut chunk = [0u8; 8192];
|
||||
match timeout(self.timeout_dur, self.stream.read(&mut chunk)).await {
|
||||
Err(_) => Err(TransportError::Timeout),
|
||||
Ok(Err(e)) => {
|
||||
// 与 perform_request 相同的良性拆连接处理。
|
||||
// Same benign-teardown handling as perform_request.
|
||||
let benign = matches!(
|
||||
e.kind(),
|
||||
std::io::ErrorKind::UnexpectedEof | std::io::ErrorKind::ConnectionReset
|
||||
);
|
||||
if benign && !self.pending.is_empty() {
|
||||
Ok(None)
|
||||
} else {
|
||||
Err(TransportError::Io(format!("read response body: {}", e)))
|
||||
}
|
||||
}
|
||||
Ok(Ok(0)) => Ok(None),
|
||||
Ok(Ok(n)) => Ok(Some(chunk[..n].to_vec())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ChunkStream for StreamingBody {
|
||||
fn next_chunk<'a>(
|
||||
&'a mut self,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = TransportResult<Option<Vec<u8>>>> + Send + 'a>>
|
||||
{
|
||||
Box::pin(async move { self.next_chunk_impl().await })
|
||||
}
|
||||
}
|
||||
|
||||
impl StreamingBody {
|
||||
async fn next_chunk_impl(&mut self) -> TransportResult<Option<Vec<u8>>> {
|
||||
if self.finished {
|
||||
return Ok(None);
|
||||
}
|
||||
loop {
|
||||
match self.mode {
|
||||
BodyLength::Fixed(_) => {
|
||||
if self.remaining == 0 {
|
||||
self.finished = true;
|
||||
return Ok(None);
|
||||
}
|
||||
if !self.pending.is_empty() {
|
||||
let take = self.pending.len().min(self.remaining);
|
||||
let out: Vec<u8> = self.pending.drain(..take).collect();
|
||||
self.remaining -= take;
|
||||
return Ok(Some(out));
|
||||
}
|
||||
match self.read_raw().await? {
|
||||
Some(chunk) => self.pending.extend_from_slice(&chunk),
|
||||
None => {
|
||||
// 提前 EOF：响应体被截断。
|
||||
// Premature EOF: the body was truncated.
|
||||
if self.remaining > 0 {
|
||||
return Err(TransportError::Io("body truncated".into()));
|
||||
}
|
||||
self.finished = true;
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
}
|
||||
BodyLength::Chunked => {
|
||||
// 先取解码器现有输出；无输出且未结束才读套接字（避免跨 await 持有借用）。
|
||||
// Drain the decoder's buffered output first; only read the
|
||||
// socket when there is none and the stream is not finished
|
||||
// (avoids holding a borrow across an await).
|
||||
if let Some(out) = self.decoder.as_mut().and_then(|d| d.take_decoded()) {
|
||||
return Ok(Some(out));
|
||||
}
|
||||
if self
|
||||
.decoder
|
||||
.as_ref()
|
||||
.map(|d| d.is_finished())
|
||||
.unwrap_or(false)
|
||||
{
|
||||
self.finished = true;
|
||||
return Ok(None);
|
||||
}
|
||||
match self.read_raw().await? {
|
||||
Some(chunk) => {
|
||||
if let Some(d) = self.decoder.as_mut() {
|
||||
d.feed(&chunk);
|
||||
}
|
||||
}
|
||||
None => {
|
||||
let finished = self
|
||||
.decoder
|
||||
.as_ref()
|
||||
.map(|d| d.is_finished())
|
||||
.unwrap_or(false);
|
||||
if !finished {
|
||||
return Err(TransportError::Io("chunked body truncated".into()));
|
||||
}
|
||||
self.finished = true;
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
}
|
||||
BodyLength::UntilClose => {
|
||||
if !self.pending.is_empty() {
|
||||
return Ok(Some(std::mem::take(&mut self.pending)));
|
||||
}
|
||||
if let Some(chunk) = self.read_raw().await? {
|
||||
return Ok(Some(chunk));
|
||||
}
|
||||
self.finished = true;
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 增量分块传输编码（chunked）解码器。
|
||||
/// An incremental chunked transfer-encoding decoder.
|
||||
///
|
||||
/// 字节以任意边界喂入（网络分块不保证与 HTTP 分块对齐），解码结果累积在
|
||||
/// 内部缓冲区中，通过 [`take_decoded`](Self::take_decoded) 取走。
|
||||
/// Bytes are fed at arbitrary boundaries (network chunks need not align with
|
||||
/// HTTP chunks); decoded bytes accumulate internally and are drained via
|
||||
/// [`take_decoded`](Self::take_decoded).
|
||||
pub(crate) struct ChunkedDecoder {
|
||||
/// 尚未消费的原始字节。
|
||||
/// Raw bytes not yet consumed.
|
||||
raw: Vec<u8>,
|
||||
/// 已消费位置。
|
||||
/// Consumed position within `raw`.
|
||||
pos: usize,
|
||||
/// 当前分块剩余数据字节数；`None` 表示正在读取分块大小行。
|
||||
/// Remaining data bytes of the current chunk; `None` while reading the size line.
|
||||
remaining: Option<usize>,
|
||||
/// 分块数据之后是否在等待尾随 CRLF。
|
||||
/// Whether we are waiting for the CRLF after chunk data.
|
||||
expect_crlf: bool,
|
||||
/// 是否已读到终止分块（大小为 0）。
|
||||
/// Whether the terminating (size-0) chunk has been read.
|
||||
finished: bool,
|
||||
/// 解码后的输出字节。
|
||||
/// Decoded output bytes.
|
||||
decoded: Vec<u8>,
|
||||
}
|
||||
|
||||
impl ChunkedDecoder {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
raw: Vec::new(),
|
||||
pos: 0,
|
||||
remaining: None,
|
||||
expect_crlf: false,
|
||||
finished: false,
|
||||
decoded: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// 喂入一块原始字节（含分块框架）。
|
||||
/// Feed a chunk of raw bytes (including chunk framing).
|
||||
fn feed(&mut self, bytes: &[u8]) {
|
||||
self.raw.extend_from_slice(bytes);
|
||||
self.decode();
|
||||
}
|
||||
|
||||
/// 取走当前已解码的全部字节。
|
||||
/// Drain all currently decoded bytes.
|
||||
fn take_decoded(&mut self) -> Option<Vec<u8>> {
|
||||
if self.decoded.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(std::mem::take(&mut self.decoded))
|
||||
}
|
||||
}
|
||||
|
||||
/// 是否已到达流末尾（读到大小为 0 的终止分块）。
|
||||
/// Whether the end of the stream (size-0 terminator) has been reached.
|
||||
fn is_finished(&self) -> bool {
|
||||
self.finished
|
||||
}
|
||||
|
||||
/// 尽可能多地解码当前原始缓冲区。
|
||||
/// Decode as much of the raw buffer as possible.
|
||||
fn decode(&mut self) {
|
||||
loop {
|
||||
if self.finished {
|
||||
return;
|
||||
}
|
||||
if let Some(rem) = self.remaining {
|
||||
if rem == 0 {
|
||||
// 消费分块数据后的尾随 CRLF（或单独的 LF）。
|
||||
// Consume the trailing CRLF (or bare LF) after chunk data.
|
||||
if self.expect_crlf {
|
||||
if self.pos + 2 <= self.raw.len()
|
||||
&& self.raw[self.pos] == b'\r'
|
||||
&& self.raw[self.pos + 1] == b'\n'
|
||||
{
|
||||
self.pos += 2;
|
||||
self.expect_crlf = false;
|
||||
self.remaining = None;
|
||||
continue;
|
||||
}
|
||||
if self.pos < self.raw.len() && self.raw[self.pos] == b'\n' {
|
||||
self.pos += 1;
|
||||
self.expect_crlf = false;
|
||||
self.remaining = None;
|
||||
continue;
|
||||
}
|
||||
return; // 需要更多数据 / need more data
|
||||
}
|
||||
} else {
|
||||
let avail = self.raw.len() - self.pos;
|
||||
if avail == 0 {
|
||||
return; // 需要更多数据 / need more data
|
||||
}
|
||||
let take = avail.min(rem);
|
||||
self.decoded
|
||||
.extend_from_slice(&self.raw[self.pos..self.pos + take]);
|
||||
self.pos += take;
|
||||
self.remaining = Some(rem - take);
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
// 读取分块大小行（以 '\n' 结束，可带分块扩展 ";..."）。
|
||||
// Read the chunk size line (terminated by '\n', may carry ";...").
|
||||
let tail = &self.raw[self.pos..];
|
||||
let nl = tail.iter().position(|&b| b == b'\n');
|
||||
match nl {
|
||||
None => return, // 需要更多数据 / need more data
|
||||
Some(nl) => {
|
||||
let line = &tail[..nl];
|
||||
let size_str = std::str::from_utf8(line)
|
||||
.unwrap_or("")
|
||||
.trim_end_matches('\r');
|
||||
let size_hex = size_str.split(';').next().unwrap_or("0").trim();
|
||||
let size = usize::from_str_radix(size_hex, 16).unwrap_or(0);
|
||||
self.pos += nl + 1; // 消费该行及其 '\n' / consume the line + '\n'
|
||||
if size == 0 {
|
||||
self.finished = true;
|
||||
return;
|
||||
}
|
||||
self.remaining = Some(size);
|
||||
self.expect_crlf = true;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 解码 HTTP/1.1 分块传输编码（chunked）的响应体。
|
||||
/// Decode an HTTP/1.1 chunked transfer-encoded body.
|
||||
fn decode_chunked(bytes: &[u8]) -> TransportResult<Vec<u8>> {
|
||||
pub fn decode_chunked(bytes: &[u8]) -> TransportResult<Vec<u8>> {
|
||||
let mut out = Vec::new();
|
||||
let mut pos = 0;
|
||||
while pos < bytes.len() {
|
||||
|
|
@ -177,29 +597,84 @@ fn find_line_end(bytes: &[u8]) -> TransportResult<usize> {
|
|||
.ok_or_else(|| TransportError::Io("missing chunk line terminator".into()))
|
||||
}
|
||||
|
||||
/// 在 `haystack` 中查找 `needle` 首次出现的字节偏移。
|
||||
/// Find the first byte offset of `needle` within `haystack`.
|
||||
fn find_subsequence(haystack: &[u8], needle: &[u8]) -> Option<usize> {
|
||||
haystack
|
||||
.windows(needle.len())
|
||||
.position(|window| window == needle)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn decodes_simple_chunked_body() {
|
||||
// 两个分块："Hello" 与 " World"，随后是终止分块。
|
||||
// Two chunks: "Hello" and " World", then terminating chunk.
|
||||
let body = b"5\r\nHello\r\n6\r\n World\r\n0\r\n\r\n";
|
||||
let decoded = decode_chunked(body).unwrap();
|
||||
assert_eq!(decoded, b"Hello World");
|
||||
/// 把所有解码输出合并为一个字符串（测试辅助）。
|
||||
/// Concatenate all decoded output into one string (test helper).
|
||||
fn drain(dec: &mut ChunkedDecoder) -> String {
|
||||
let mut out = Vec::new();
|
||||
loop {
|
||||
match dec.take_decoded() {
|
||||
Some(mut chunk) => out.append(&mut chunk),
|
||||
None => {
|
||||
if dec.is_finished() {
|
||||
break;
|
||||
}
|
||||
return String::from_utf8_lossy(&out).to_string();
|
||||
}
|
||||
}
|
||||
}
|
||||
String::from_utf8_lossy(&out).to_string()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_chunk_with_extension() {
|
||||
let body = b"5;name=value\r\nHello\r\n0\r\n\r\n";
|
||||
let decoded = decode_chunked(body).unwrap();
|
||||
assert_eq!(decoded, b"Hello");
|
||||
fn decodes_single_chunk_fed_at_once() {
|
||||
let mut dec = ChunkedDecoder::new();
|
||||
dec.feed(b"5\r\nhello\r\n0\r\n\r\n");
|
||||
assert_eq!(drain(&mut dec), "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_single_chunk() {
|
||||
let body = b"3\r\nabc\r\n0\r\n\r\n";
|
||||
assert_eq!(decode_chunked(body).unwrap(), b"abc");
|
||||
fn decodes_chunk_split_across_feeds() {
|
||||
// 网络分块与 HTTP 分块不对齐：一次喂一半。
|
||||
// Network chunks misalign with HTTP chunks: feed half at a time.
|
||||
let mut dec = ChunkedDecoder::new();
|
||||
dec.feed(b"5\r\nhe");
|
||||
assert_eq!(drain(&mut dec), "he");
|
||||
dec.feed(b"llo\r\n0\r\n\r\n");
|
||||
assert_eq!(drain(&mut dec), "llo");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_multiple_chunks_and_extension() {
|
||||
let mut dec = ChunkedDecoder::new();
|
||||
dec.feed(b"5;name=value\r\nhello\r\n6\r\n world\r\n0\r\n\r\n");
|
||||
assert_eq!(drain(&mut dec), "hello world");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn yields_partial_data_before_terminator() {
|
||||
let mut dec = ChunkedDecoder::new();
|
||||
// 第一个分块完整、第二个分块只喂了部分，应先把已解码部分吐出。
|
||||
// First chunk complete, second partial: yield decoded bytes so far.
|
||||
dec.feed(b"3\r\nabc\r\n5\r\n12");
|
||||
assert_eq!(drain(&mut dec), "abc12");
|
||||
dec.feed(b"345\r\n0\r\n\r\n");
|
||||
assert_eq!(drain(&mut dec), "345");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chunked_streaming_body_end_to_end() {
|
||||
// 模拟完整 SSE 流（chunked 编码）经 StreamingBody 逐块吐出。
|
||||
// Simulate a full SSE stream (chunked-encoded) being drained via StreamingBody.
|
||||
// 这里只测解码器端到端；网络部分由 provider 的 mock transport 覆盖。
|
||||
// Only the decoder is exercised here; the network side is covered by the
|
||||
// providers' mock transport tests.
|
||||
let mut dec = ChunkedDecoder::new();
|
||||
let payload = "event: message_start\r\ndata: {\"type\":\"x\"}\r\n\r\n";
|
||||
assert_eq!(payload.len(), 44);
|
||||
dec.feed(format!("{:x}\r\n{}", payload.len(), payload).as_bytes());
|
||||
dec.feed(b"0\r\n\r\n");
|
||||
assert_eq!(drain(&mut dec), payload);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,59 @@
|
|||
//! HTTP/1.1 编解码的集成测试。
|
||||
//! Integration tests for the HTTP/1.1 codec.
|
||||
|
||||
use focus_transport::http::{
|
||||
body_length, parse_response_head, BodyLength, HttpRequest, HttpResponse,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[test]
|
||||
fn serializes_post_request() {
|
||||
let req = HttpRequest::post("api.example.com", "/v1/messages")
|
||||
.header("Authorization", "Bearer xyz")
|
||||
.json_body(r#"{"hello":"world"}"#);
|
||||
let wire = req.serialize();
|
||||
let text = String::from_utf8(wire).unwrap();
|
||||
assert!(text.starts_with("POST /v1/messages HTTP/1.1\r\n"));
|
||||
assert!(text.contains("Host: api.example.com\r\n"));
|
||||
assert!(text.contains("Authorization: Bearer xyz\r\n"));
|
||||
assert!(text.contains("Content-Type: application/json\r\n"));
|
||||
assert!(text.contains("Content-Length: 17\r\n"));
|
||||
assert!(text.ends_with("{\"hello\":\"world\"}"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_response_head() {
|
||||
let raw =
|
||||
b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 5\r\n\r\nhello";
|
||||
let (resp, consumed) = parse_response_head(raw).unwrap();
|
||||
assert_eq!(resp.status, 200);
|
||||
assert_eq!(
|
||||
resp.headers.get("content-type").unwrap(),
|
||||
"application/json"
|
||||
);
|
||||
assert_eq!(resp.headers.get("content-length").unwrap(), "5");
|
||||
assert_eq!(consumed, raw.len() - 5);
|
||||
assert_eq!(body_length(&resp), BodyLength::Fixed(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn detects_chunked() {
|
||||
let raw = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
|
||||
let (resp, _) = parse_response_head(raw).unwrap();
|
||||
assert_eq!(body_length(&resp), BodyLength::Chunked);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_success_for_2xx() {
|
||||
let resp = HttpResponse {
|
||||
status: 204,
|
||||
headers: HashMap::new(),
|
||||
body: vec![],
|
||||
};
|
||||
assert!(resp.is_success());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_incomplete_head() {
|
||||
assert!(parse_response_head(b"HTTP/1.1 200 OK\r\n").is_err());
|
||||
}
|
||||
|
|
@ -0,0 +1,101 @@
|
|||
//! SSE 增量解析器的集成测试。
|
||||
//! Integration tests for the incremental SSE parser.
|
||||
|
||||
use focus_transport::sse::{SseEvent, SseParser};
|
||||
|
||||
fn collect(parser: &mut SseParser) -> Vec<SseEvent> {
|
||||
let mut out = Vec::new();
|
||||
while let Some(e) = parser.next_event() {
|
||||
out.push(e);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_single_event() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: hello\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events.len(), 1);
|
||||
assert_eq!(events[0].data, "hello");
|
||||
assert_eq!(events[0].event, "message");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_event_with_type() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("event: ping\ndata: 1\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].event, "ping");
|
||||
assert_eq!(events[0].data, "1");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn joins_multiline_data() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: line1\ndata: line2\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "line1\nline2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_crlf() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: hi\r\n\r\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "hi");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ignores_comments_and_unknown_fields() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str(": this is a comment\ndata: ok\nfoo: bar\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "ok");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_id_and_retry() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("id: 42\nretry: 5000\ndata: x\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].id, "42");
|
||||
assert_eq!(events[0].retry, Some(5000));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_split_across_chunks() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("dat");
|
||||
assert!(collect(&mut p).is_empty());
|
||||
p.feed_str("a: hello\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_split_across_line_boundary() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: first\n\ndata: sec");
|
||||
assert_eq!(collect(&mut p).len(), 1);
|
||||
p.feed_str("ond\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "second");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multiple_events_in_one_chunk() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data: a\n\ndata: b\n\ndata: c\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events.len(), 3);
|
||||
assert_eq!(events[2].data, "c");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_data_line() {
|
||||
let mut p = SseParser::new();
|
||||
p.feed_str("data:\n\n");
|
||||
let events = collect(&mut p);
|
||||
assert_eq!(events[0].data, "");
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
//! transport 模块（chunked 解码）的集成测试。
|
||||
//! Integration tests for the transport module (chunked decoding).
|
||||
|
||||
use focus_transport::transport::decode_chunked;
|
||||
|
||||
#[test]
|
||||
fn decodes_simple_chunked_body() {
|
||||
// 两个分块："Hello" 与 " World"，随后是终止分块。
|
||||
// Two chunks: "Hello" and " World", then terminating chunk.
|
||||
let body = b"5\r\nHello\r\n6\r\n World\r\n0\r\n\r\n";
|
||||
let decoded = decode_chunked(body).unwrap();
|
||||
assert_eq!(decoded, b"Hello World");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_chunk_with_extension() {
|
||||
let body = b"5;name=value\r\nHello\r\n0\r\n\r\n";
|
||||
let decoded = decode_chunked(body).unwrap();
|
||||
assert_eq!(decoded, b"Hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_single_chunk() {
|
||||
let body = b"3\r\nabc\r\n0\r\n\r\n";
|
||||
assert_eq!(decode_chunked(body).unwrap(), b"abc");
|
||||
}
|
||||
Loading…
Reference in New Issue