From 7f0088615071f69c916210d8e83d14b13f1d6284 Mon Sep 17 00:00:00 2001 From: DaiChaoXiong Date: Sun, 9 Aug 2026 20:59:33 +0800 Subject: [PATCH] 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 --- crates/focus-transport/src/http.rs | 74 +-- crates/focus-transport/src/lib.rs | 2 +- crates/focus-transport/src/sse.rs | 102 ---- crates/focus-transport/src/transport.rs | 525 +++++++++++++++++- crates/focus-transport/tests/http_tests.rs | 59 ++ crates/focus-transport/tests/sse_tests.rs | 101 ++++ .../focus-transport/tests/transport_tests.rs | 26 + 7 files changed, 703 insertions(+), 186 deletions(-) create mode 100644 crates/focus-transport/tests/http_tests.rs create mode 100644 crates/focus-transport/tests/sse_tests.rs create mode 100644 crates/focus-transport/tests/transport_tests.rs diff --git a/crates/focus-transport/src/http.rs b/crates/focus-transport/src/http.rs index d8ae6a0..bc992bf 100644 --- a/crates/focus-transport/src/http.rs +++ b/crates/focus-transport/src/http.rs @@ -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 { .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()); - } -} diff --git a/crates/focus-transport/src/lib.rs b/crates/focus-transport/src/lib.rs index 64d4890..d4e3764 100644 --- a/crates/focus-transport/src/lib.rs +++ b/crates/focus-transport/src/lib.rs @@ -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}; diff --git a/crates/focus-transport/src/sse.rs b/crates/focus-transport/src/sse.rs index a6582c3..c838b67 100644 --- a/crates/focus-transport/src/sse.rs +++ b/crates/focus-transport/src/sse.rs @@ -192,105 +192,3 @@ impl SseParser { } } } - -#[cfg(test)] -mod tests { - use super::*; - - fn collect(parser: &mut SseParser) -> Vec { - 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, ""); - } -} diff --git a/crates/focus-transport/src/transport.rs b/crates/focus-transport/src/transport.rs index c9d9669..8054f7d 100644 --- a/crates/focus-transport/src/transport.rs +++ b/crates/focus-transport/src/transport.rs @@ -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>>> + Send + 'a>>; +/// 普通请求的 future 类型。 +/// Future type for a plain request. +pub type RequestFuture<'a> = + Pin> + Send + 'a>>; +/// 流式请求的 future 类型。 +/// Future type for a streaming request. +pub type StreamRequestFuture<'a> = + Pin>> + 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> + 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> + 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 { - 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 { 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> { + 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 = 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, + /// 固定长度模式下剩余字节数。 + /// Remaining bytes in fixed-length mode. + remaining: usize, + /// 分块模式下使用的增量解码器。 + /// Incremental decoder used in chunked mode. + decoder: Option, + finished: bool, + timeout_dur: Duration, +} + +impl StreamingBody { + fn new( + stream: crate::tls::TlsConnection, + mode: BodyLength, + pending: Vec, + 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>> { + 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>>> + Send + 'a>> + { + Box::pin(async move { self.next_chunk_impl().await }) + } +} + +impl StreamingBody { + async fn next_chunk_impl(&mut self) -> TransportResult>> { + 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 = 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, + /// 已消费位置。 + /// Consumed position within `raw`. + pos: usize, + /// 当前分块剩余数据字节数；`None` 表示正在读取分块大小行。 + /// Remaining data bytes of the current chunk; `None` while reading the size line. + remaining: Option, + /// 分块数据之后是否在等待尾随 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, +} + +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> { + 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> { +pub fn decode_chunked(bytes: &[u8]) -> TransportResult> { let mut out = Vec::new(); let mut pos = 0; while pos < bytes.len() { @@ -177,29 +597,84 @@ fn find_line_end(bytes: &[u8]) -> TransportResult { .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 { + 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); } } diff --git a/crates/focus-transport/tests/http_tests.rs b/crates/focus-transport/tests/http_tests.rs new file mode 100644 index 0000000..ebea87a --- /dev/null +++ b/crates/focus-transport/tests/http_tests.rs @@ -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()); +} diff --git a/crates/focus-transport/tests/sse_tests.rs b/crates/focus-transport/tests/sse_tests.rs new file mode 100644 index 0000000..037cf65 --- /dev/null +++ b/crates/focus-transport/tests/sse_tests.rs @@ -0,0 +1,101 @@ +//! SSE 增量解析器的集成测试。 +//! Integration tests for the incremental SSE parser. + +use focus_transport::sse::{SseEvent, SseParser}; + +fn collect(parser: &mut SseParser) -> Vec { + 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, ""); +} diff --git a/crates/focus-transport/tests/transport_tests.rs b/crates/focus-transport/tests/transport_tests.rs new file mode 100644 index 0000000..a5a99b6 --- /dev/null +++ b/crates/focus-transport/tests/transport_tests.rs @@ -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"); +}