kode_bridge/
http_client.rs

1use crate::errors::{KodeBridgeError, Result};
2use crate::parser_cache::global_parser_cache;
3use bytes::{BufMut as _, Bytes, BytesMut};
4use http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Version};
5use serde::{de::DeserializeOwned, Serialize};
6use serde_json::Value;
7use std::collections::HashMap;
8use std::str::FromStr as _;
9use std::time::Duration;
10use tokio::io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader};
11use tokio::io::{AsyncRead, AsyncWrite};
12use tracing::{debug, trace};
13
14/// Enhanced HTTP response with rich functionality
15#[derive(Debug, Clone)]
16pub struct Response {
17    status: StatusCode,
18    version: Version,
19    headers: HeaderMap,
20    body: Bytes,
21}
22
23impl Response {
24    pub const fn new(status: StatusCode, version: Version, headers: HeaderMap, body: Bytes) -> Self {
25        Self {
26            status,
27            version,
28            headers,
29            body,
30        }
31    }
32
33    /// Get HTTP status code
34    pub const fn status(&self) -> StatusCode {
35        self.status
36    }
37
38    /// Get status code as u16
39    pub const fn status_code(&self) -> u16 {
40        self.status.as_u16()
41    }
42
43    /// Get HTTP version
44    pub const fn version(&self) -> Version {
45        self.version
46    }
47
48    /// Get response headers
49    pub const fn headers(&self) -> &HeaderMap {
50        &self.headers
51    }
52
53    /// Get response body as bytes
54    pub const fn body(&self) -> &Bytes {
55        &self.body
56    }
57
58    /// Get response body as string
59    pub fn text(&self) -> Result<String> {
60        String::from_utf8(self.body.to_vec()).map_err(KodeBridgeError::from)
61    }
62
63    /// Parse response body as JSON
64    pub fn json<T>(&self) -> Result<T>
65    where
66        T: DeserializeOwned,
67    {
68        serde_json::from_slice(&self.body).map_err(KodeBridgeError::from)
69    }
70
71    /// Parse response body as generic JSON value
72    pub fn json_value(&self) -> Result<Value> {
73        serde_json::from_slice(&self.body).map_err(KodeBridgeError::from)
74    }
75
76    /// Check if response indicates success (2xx status)
77    pub fn is_success(&self) -> bool {
78        self.status.is_success()
79    }
80
81    /// Check if response indicates client error (4xx status)
82    pub fn is_client_error(&self) -> bool {
83        self.status.is_client_error()
84    }
85
86    /// Check if response indicates server error (5xx status)
87    pub fn is_server_error(&self) -> bool {
88        self.status.is_server_error()
89    }
90
91    /// Check if response indicates redirection (3xx status)
92    pub fn is_redirection(&self) -> bool {
93        self.status.is_redirection()
94    }
95
96    /// Get content length from headers
97    pub fn content_length(&self) -> Option<u64> {
98        self.headers
99            .get(header::CONTENT_LENGTH)?
100            .to_str()
101            .ok()?
102            .parse()
103            .ok()
104    }
105
106    /// Get content type from headers
107    pub fn content_type(&self) -> Option<&str> {
108        self.headers.get(header::CONTENT_TYPE)?.to_str().ok()
109    }
110
111    /// Convert to legacy Response format for backward compatibility
112    pub fn to_legacy(&self) -> crate::response::LegacyResponse {
113        let headers_map: HashMap<String, String> = self
114            .headers
115            .iter()
116            .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
117            .collect();
118
119        crate::response::LegacyResponse {
120            status: self.status.as_u16(),
121            headers: serde_json::to_value(headers_map).unwrap_or(Value::Null),
122            body: String::from_utf8_lossy(&self.body).to_string(),
123        }
124    }
125}
126
127/// HTTP request builder with fluent interface
128#[derive(Debug)]
129pub struct RequestBuilder {
130    method: Method,
131    uri: String,
132    headers: HeaderMap,
133    body: Option<Bytes>,
134}
135
136impl RequestBuilder {
137    pub fn new(method: Method, uri: String) -> Self {
138        let mut headers = HeaderMap::new();
139        headers.insert(header::HOST, HeaderValue::from_static("localhost"));
140        headers.insert(header::USER_AGENT, HeaderValue::from_static("kode-bridge/0.1"));
141
142        Self {
143            method,
144            uri,
145            headers,
146            body: None,
147        }
148    }
149
150    /// Add a custom header
151    pub fn header(mut self, key: &str, value: &str) -> Self {
152        if let (Ok(name), Ok(val)) = (HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_str(value)) {
153            self.headers.insert(name, val);
154        }
155        self
156    }
157
158    /// Set a pre-serialized body and content type.
159    pub fn body_bytes(mut self, body: Bytes, content_type: &'static str) -> Result<Self> {
160        self.headers
161            .insert(header::CONTENT_TYPE, HeaderValue::from_static(content_type));
162
163        let content_length = body.len().to_string();
164        self.headers.insert(
165            header::CONTENT_LENGTH,
166            HeaderValue::from_str(&content_length).map_err(|e| KodeBridgeError::Http(e.into()))?,
167        );
168
169        self.body = Some(body);
170        Ok(self)
171    }
172
173    /// Set JSON body with optimized serialization
174    pub fn json<T>(self, body: &T) -> Result<Self>
175    where
176        T: Serialize,
177    {
178        let mut buffer = BytesMut::with_capacity(1024);
179        {
180            let writer = (&mut buffer).writer();
181            serde_json::to_writer(writer, body).map_err(KodeBridgeError::from)?;
182        }
183
184        self.body_bytes(buffer.freeze(), "application/json")
185    }
186
187    /// Build the HTTP request as bytes with optimized allocation
188    pub fn build(self) -> Result<Bytes> {
189        let mut request = BytesMut::with_capacity(1024 + self.body.as_ref().map(|b| b.len()).unwrap_or(0));
190
191        // Request line - 使用更高效的写入方式
192        request.extend_from_slice(self.method.as_str().as_bytes());
193        request.extend_from_slice(b" ");
194        request.extend_from_slice(self.uri.as_bytes());
195        request.extend_from_slice(b" HTTP/1.1\r\n");
196
197        // Headers - 批量写入减少系统调用
198        let mut headers_buffer = BytesMut::with_capacity(512);
199        for (key, value) in &self.headers {
200            headers_buffer.extend_from_slice(key.as_str().as_bytes());
201            headers_buffer.extend_from_slice(b": ");
202            headers_buffer.extend_from_slice(value.as_bytes());
203            headers_buffer.extend_from_slice(b"\r\n");
204        }
205        request.extend_from_slice(&headers_buffer);
206
207        // End of headers
208        request.extend_from_slice(b"\r\n");
209
210        // Body
211        if let Some(body) = self.body {
212            request.extend_from_slice(&body);
213        }
214
215        Ok(request.freeze())
216    }
217}
218
219/// Parse HTTP response from a stream with optimized parsing
220pub async fn parse_response<S>(stream: S) -> Result<Response>
221where
222    S: AsyncRead + Unpin + Send,
223{
224    let mut reader = BufReader::new(stream);
225
226    // 使用预分配的缓冲区避免多次分配
227    let mut headers_buffer = BytesMut::with_capacity(1024);
228
229    // 先读取状态行
230    let mut status_line = String::new();
231    if reader.read_line(&mut status_line).await? == 0 {
232        return Err(KodeBridgeError::StreamClosed);
233    }
234    headers_buffer.extend_from_slice(status_line.as_bytes());
235
236    // 读取HTTP头部直到遇到空行
237    let mut line = String::new();
238    loop {
239        line.clear();
240        let bytes_read = reader.read_line(&mut line).await?;
241        if bytes_read == 0 {
242            break;
243        }
244
245        // 检查是否是头部结束（空行）
246        if line == "\r\n" || line == "\n" {
247            break;
248        }
249
250        // 将头部行添加到缓冲区
251        headers_buffer.extend_from_slice(line.as_bytes());
252    }
253
254    // 添加最后的 \r\n 来标记头部结束
255    headers_buffer.extend_from_slice(b"\r\n");
256
257    // Use cached parser for better performance
258    let mut parser = global_parser_cache().get();
259    let (status, parsed_headers) = parser
260        .parse_response(&headers_buffer)
261        .map_err(|e| match e {
262            httparse::Error::TooManyHeaders => {
263                KodeBridgeError::protocol("Too many HTTP headers in response (limit: 64)")
264            }
265            _ => KodeBridgeError::protocol(format!("Failed to parse HTTP response: {:?}", e)),
266        })?;
267
268    // Build HeaderMap
269    let mut header_map = HeaderMap::new();
270    for (name, value) in parsed_headers {
271        let header_name = HeaderName::from_str(&name).map_err(|e| KodeBridgeError::Http(e.into()))?;
272        let header_value = HeaderValue::from_str(&value).map_err(|e| KodeBridgeError::Http(e.into()))?;
273        header_map.insert(header_name, header_value);
274    }
275
276    // Determine body length
277    let content_length = header_map
278        .get(header::CONTENT_LENGTH)
279        .and_then(|v| v.to_str().ok())
280        .and_then(|s| s.parse::<usize>().ok());
281
282    let is_chunked = header_map
283        .get(header::TRANSFER_ENCODING)
284        .and_then(|v| v.to_str().ok())
285        .map(|s| s.eq_ignore_ascii_case("chunked"))
286        .unwrap_or(false);
287
288    // Read body with optimized handling for empty responses (common in PUT/POST)
289    let body = if is_chunked {
290        read_chunked_body(&mut reader).await?
291    } else if let Some(len) = content_length {
292        if len == 0 {
293            // Empty response, return directly - optimize for PUT responses
294            Bytes::new()
295        } else if len > 10 * 1024 * 1024 {
296            // Use streaming for very large responses (>10MB)
297            return Err(KodeBridgeError::protocol("Response body too large for memory"));
298        } else {
299            read_fixed_body(&mut reader, len).await?
300        }
301    } else {
302        // For responses without Content-Length, use adaptive reading with shorter timeouts for PUT
303        read_until_end_adaptive(&mut reader).await?
304    };
305
306    Ok(Response::new(
307        StatusCode::from_u16(status)?,
308        Version::HTTP_11,
309        header_map,
310        body,
311    ))
312}
313
314async fn read_chunked_body<R>(reader: &mut BufReader<R>) -> Result<Bytes>
315where
316    R: AsyncRead + Unpin + Send,
317{
318    let mut body_buffer = BytesMut::with_capacity(8192);
319
320    loop {
321        // Read chunk size line
322        let mut size_line = String::new();
323        reader.read_line(&mut size_line).await?;
324
325        let size_line = size_line.trim();
326        if size_line.is_empty() {
327            continue;
328        }
329
330        // Parse chunk size (hex)
331        let chunk_size =
332            usize::from_str_radix(size_line, 16).map_err(|_| KodeBridgeError::protocol("Invalid chunk size"))?;
333
334        if chunk_size == 0 {
335            // Last chunk, read final CRLF
336            let mut final_line = String::new();
337            reader.read_line(&mut final_line).await?;
338            break;
339        }
340
341        // Read chunk data
342        let mut chunk = vec![0u8; chunk_size];
343        reader.read_exact(&mut chunk).await?;
344        body_buffer.extend_from_slice(&chunk);
345
346        // Read trailing CRLF
347        let mut crlf = [0u8; 2];
348        reader.read_exact(&mut crlf).await?;
349    }
350
351    Ok(body_buffer.freeze())
352}
353
354async fn read_fixed_body<R>(reader: &mut BufReader<R>, len: usize) -> Result<Bytes>
355where
356    R: AsyncRead + Unpin + Send,
357{
358    let mut body = vec![0u8; len];
359    reader.read_exact(&mut body).await?;
360    Ok(Bytes::from(body))
361}
362
363async fn read_until_end_adaptive<R>(reader: &mut BufReader<R>) -> Result<Bytes>
364where
365    R: AsyncRead + Unpin + Send,
366{
367    let mut body_buffer = BytesMut::with_capacity(8192);
368    let mut read_buffer = [0u8; 512];
369    let mut consecutive_empty_reads = 0;
370
371    loop {
372        let timeout_duration = Duration::from_millis(25 + (consecutive_empty_reads * 25));
373
374        match tokio::time::timeout(timeout_duration, reader.read(&mut read_buffer)).await {
375            Ok(Ok(0)) => {
376                // EOF reached
377                break;
378            }
379            Ok(Ok(n)) => {
380                body_buffer.extend_from_slice(&read_buffer[..n]);
381                consecutive_empty_reads = 0;
382            }
383            Ok(Err(e)) => {
384                return Err(KodeBridgeError::from(e));
385            }
386            Err(_) => {
387                // Timeout occurred
388                consecutive_empty_reads += 1;
389                if consecutive_empty_reads >= 2 {
390                    // 减少到2次连续超时就结束，提高响应速度
391                    break;
392                }
393                continue;
394            }
395        }
396
397        // Safety limit to prevent unbounded memory usage - 增加到20MB
398        if body_buffer.len() > 20 * 1024 * 1024 {
399            return Err(KodeBridgeError::protocol("Response body too large"));
400        }
401    }
402
403    // Convert pooled buffer to Bytes for return
404    Ok(body_buffer.freeze())
405}
406
407/// Send HTTP request and parse response
408pub async fn send_request<S>(mut stream: S, request: Bytes) -> Result<Response>
409where
410    S: AsyncRead + AsyncWrite + Unpin + Send,
411{
412    // Send request
413    stream.write_all(&request).await?;
414    stream.flush().await?;
415
416    trace!("Sent HTTP request ({} bytes)", request.len());
417
418    // Parse response
419    let response = parse_response(stream).await?;
420
421    debug!(
422        "Received HTTP response: {} {}",
423        response.status(),
424        response.content_length().unwrap_or(0)
425    );
426
427    Ok(response)
428}
429
430#[cfg(test)]
431mod tests {
432    use super::*;
433
434    #[tokio::test]
435    async fn closed_response_stream_is_distinct_from_a_malformed_status() {
436        let (reader, writer) = tokio::io::duplex(1);
437        drop(writer);
438        assert!(matches!(
439            parse_response(reader).await,
440            Err(KodeBridgeError::StreamClosed)
441        ));
442
443        let (mut writer, reader) = tokio::io::duplex(64);
444        writer
445            .write_all(b"not an HTTP status\r\n\r\n")
446            .await
447            .unwrap();
448        drop(writer);
449        assert!(matches!(
450            parse_response(reader).await,
451            Err(KodeBridgeError::Protocol { .. })
452        ));
453    }
454}
kode_bridge/http_client.rs

kode_bridge/
http_client.rs
