kode_bridge/
ipc_http_client.rs

1use std::path::Path;
2use std::time::Duration;
3
4use serde::de::DeserializeOwned;
5use serde_json::Value;
6
7use crate::errors::{KodeBridgeError, Result};
8use crate::http_client::{send_request, RequestBuilder, Response};
9use crate::metrics::global_metrics;
10use crate::pool::{ConnectionPool, PoolConfig, PooledConnection};
11use crate::retry::{RetryConfig, RetryExecutor};
12use crate::transport::{Endpoint, IpcStream};
13use bytes::Bytes;
14use http::Method;
15use std::str::FromStr as _;
16use tracing::{debug, trace};
17
18/// Configuration for IPC HTTP client
19#[derive(Debug, Clone)]
20pub struct ClientConfig {
21    /// Default timeout for requests
22    pub default_timeout: Duration,
23    /// Connection pool configuration
24    pub pool_config: PoolConfig,
25    /// Enable connection pooling
26    pub enable_pooling: bool,
27    /// Retry configuration
28    pub max_retries: usize,
29    pub retry_delay: Duration,
30    /// Concurrent request limit
31    pub max_concurrent_requests: usize,
32    /// Rate limiting: max requests per second
33    pub max_requests_per_second: Option<f64>,
34}
35
36impl Default for ClientConfig {
37    fn default() -> Self {
38        Self {
39            default_timeout: Duration::from_secs(5), // 减少默认超时到5秒
40            pool_config: PoolConfig::default(),
41            enable_pooling: true,
42            max_retries: 3,                         // 减少重试次数
43            retry_delay: Duration::from_millis(25), // 减少重试延迟
44            max_concurrent_requests: 16,            // 增加并发请求数
45            max_requests_per_second: Some(50.0),    // 增加请求速率限制
46        }
47    }
48}
49
50/// Generic IPC HTTP client that works on both Unix and Windows platforms
51///
52/// This client is optimized for request-response patterns with connection pooling support.
53/// For streaming functionality, use `IpcStreamClient` instead.
54pub struct IpcHttpClient {
55    endpoint: Endpoint,
56    config: ClientConfig,
57    pool: Option<ConnectionPool>,
58    retry_executor: RetryExecutor,
59    /// 专门用于PUT请求的重试执行器
60    put_retry_executor: RetryExecutor,
61}
62
63/// HTTP request builder for fluent API
64pub struct HttpRequestBuilder<'a> {
65    client: &'a IpcHttpClient,
66    method: Method,
67    path: String,
68    body: Option<RequestBody>,
69    timeout: Option<Duration>,
70    headers: Vec<(String, String)>,
71    /// PUT专用优化标志
72    put_optimized: bool,
73    /// 预期数据大小，用于选择合适的缓冲区和超时
74    expected_size: Option<usize>,
75}
76
77enum RequestBody {
78    Json(Value),
79    JsonBytes(Bytes),
80}
81
82/// Enhanced HTTP response wrapper with chainable methods
83#[derive(Debug)]
84pub struct HttpResponse {
85    inner: Response,
86}
87
88impl HttpResponse {
89    const fn new(response: Response) -> Self {
90        Self { inner: response }
91    }
92
93    /// Get the HTTP status code
94    pub const fn status(&self) -> u16 {
95        self.inner.status_code()
96    }
97
98    /// Get response headers as JSON value (for backward compatibility)
99    pub fn headers(&self) -> Value {
100        let headers_map: std::collections::HashMap<String, String> = self
101            .inner
102            .headers()
103            .iter()
104            .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
105            .collect();
106        serde_json::to_value(headers_map).unwrap_or(Value::Null)
107    }
108
109    /// Get response body as string
110    pub fn body(&self) -> Result<String> {
111        self.inner.text()
112    }
113
114    /// Check if response indicates success (2xx status)
115    pub fn is_success(&self) -> bool {
116        self.inner.is_success()
117    }
118
119    /// Check if response indicates client error (4xx status)
120    pub fn is_client_error(&self) -> bool {
121        self.inner.is_client_error()
122    }
123
124    /// Check if response indicates server error (5xx status)
125    pub fn is_server_error(&self) -> bool {
126        self.inner.is_server_error()
127    }
128
129    /// Get content length from headers
130    pub fn content_length(&self) -> u64 {
131        self.inner.content_length().unwrap_or(0)
132    }
133
134    /// Parse response body as JSON
135    pub fn json<T>(&self) -> Result<T>
136    where
137        T: DeserializeOwned,
138    {
139        self.inner.json()
140    }
141
142    /// Parse response body as generic JSON value
143    pub fn json_value(&self) -> Result<Value> {
144        self.inner.json_value()
145    }
146
147    /// Get the underlying modern Response
148    pub fn into_inner(self) -> Response {
149        self.inner
150    }
151
152    /// Convert to legacy Response format for backward compatibility
153    pub fn to_legacy(&self) -> crate::response::LegacyResponse {
154        self.inner.to_legacy()
155    }
156}
157
158impl IpcHttpClient {
159    /// Create a new IPC HTTP client with default configuration
160    pub fn new<P>(path: P) -> Result<Self>
161    where
162        P: AsRef<Path>,
163    {
164        Self::with_config(path, ClientConfig::default())
165    }
166
167    /// Create a new IPC HTTP client with custom configuration
168    pub fn with_config<P>(path: P, config: ClientConfig) -> Result<Self>
169    where
170        P: AsRef<Path>,
171    {
172        let endpoint = Endpoint::new(path)?;
173
174        let pool = if config.enable_pooling {
175            Some(ConnectionPool::new(endpoint.clone(), config.pool_config.clone()))
176        } else {
177            None
178        };
179
180        // Create retry executor with optimized configuration for different request types
181        let retry_config = RetryConfig::for_network_operations()
182            .max_attempts(config.max_retries)
183            .base_delay(Duration::from_millis(config.pool_config.retry_delay_ms));
184
185        let retry_executor = RetryExecutor::new(retry_config);
186
187        // 创建专门用于PUT请求的快速重试执行器
188        let put_retry_config = RetryConfig::for_put_requests();
189        let put_retry_executor = RetryExecutor::new(put_retry_config);
190
191        Ok(Self {
192            endpoint,
193            config,
194            pool,
195            retry_executor,
196            put_retry_executor,
197        })
198    }
199
200    /// Create a direct connection (bypassing pool)
201    async fn create_direct_connection(&self) -> Result<IpcStream> {
202        let mut last_error = None;
203
204        for attempt in 0..self.config.max_retries {
205            if attempt > 0 {
206                tokio::time::sleep(self.config.retry_delay).await;
207            }
208
209            match IpcStream::connect(&self.endpoint).await {
210                Ok(stream) => {
211                    debug!("Created direct connection on attempt {}", attempt + 1);
212                    return Ok(stream);
213                }
214                Err(e) => {
215                    trace!("Connection attempt {} failed: {}", attempt + 1, e);
216                    last_error = Some(e);
217                }
218            }
219        }
220
221        Err(KodeBridgeError::connection(format!(
222            "Failed to create connection after {} attempts: {}",
223            self.config.max_retries,
224            last_error
225                .map(|e| e.to_string())
226                .unwrap_or_else(|| "Unknown error".to_string())
227        )))
228    }
229
230    /// Get a connection (from pool or create new)
231    async fn get_connection(&self) -> Result<Either<PooledConnection, IpcStream>> {
232        let metrics = global_metrics();
233
234        if let Some(ref pool) = self.pool {
235            match pool.get_connection().await {
236                Ok(conn) => {
237                    metrics.connection_created(true); // From pool
238                    Ok(Either::Pool(conn))
239                }
240                Err(e) => {
241                    metrics.connection_failed();
242                    Err(e)
243                }
244            }
245        } else {
246            match self.create_direct_connection().await {
247                Ok(stream) => {
248                    metrics.connection_created(false); // Direct connection
249                    Ok(Either::Direct(stream))
250                }
251                Err(e) => {
252                    metrics.connection_failed();
253                    Err(e)
254                }
255            }
256        }
257    }
258
259    /// Legacy request method for backward compatibility
260    pub async fn request(
261        &self,
262        method: &str,
263        path: &str,
264        body: Option<&serde_json::Value>,
265    ) -> crate::errors::AnyResult<crate::response::LegacyResponse> {
266        let response = self
267            .send_request_internal(method, path, body, self.config.default_timeout)
268            .await?;
269        Ok(response.to_legacy())
270    }
271
272    /// Internal method to send requests with enhanced retry logic
273    async fn send_request_internal(
274        &self,
275        method: &str,
276        path: &str,
277        body: Option<&Value>,
278        timeout: Duration,
279    ) -> Result<Response> {
280        let method =
281            Method::from_str(method).map_err(|e| KodeBridgeError::invalid_request(format!("Invalid method: {}", e)))?;
282
283        let mut builder = RequestBuilder::new(method.clone(), path.to_string());
284
285        // Note: This method doesn't support custom headers for backward compatibility
286        // Use the fluent API (get(), post(), etc.) for custom headers
287
288        if let Some(json_body) = body {
289            builder = builder.json(json_body)?;
290        }
291
292        let request = builder.build()?;
293
294        // Use smart retry mechanism
295        self.retry_executor
296            .execute_with_context(&format!("{} {}", method.as_str(), path), || async {
297                // Execute with timeout
298                let result = tokio::time::timeout(timeout, async {
299                    let mut connection = self.get_connection().await?;
300
301                    match &mut connection {
302                        Either::Pool(conn) => {
303                            if let Some(stream) = conn.stream() {
304                                let result = send_request(stream, request.clone()).await;
305                                if result.is_err() {
306                                    conn.invalidate();
307                                }
308                                result
309                            } else {
310                                conn.invalidate();
311                                Err(KodeBridgeError::connection("Pooled connection is invalid"))
312                            }
313                        }
314                        Either::Direct(stream) => send_request(stream, request.clone()).await,
315                    }
316                })
317                .await;
318
319                match result {
320                    Ok(response) => response,
321                    Err(_) => Err(KodeBridgeError::timeout(timeout.as_millis() as u64)),
322                }
323            })
324            .await
325    }
326
327    /// Enhanced request sending with PUT optimization support
328    async fn send_request_with_optimization(
329        &self,
330        method: &str,
331        path: &str,
332        body: Option<&RequestBody>,
333        headers: &[(String, String)],
334        timeout: Duration,
335        is_put_optimized: bool,
336        expected_size: Option<usize>,
337    ) -> Result<Response> {
338        let method_enum =
339            Method::from_str(method).map_err(|e| KodeBridgeError::invalid_request(format!("Invalid method: {}", e)))?;
340
341        let mut builder = RequestBuilder::new(method_enum.clone(), path.to_string());
342
343        // Add custom headers
344        for (key, value) in headers {
345            builder = builder.header(key.as_str(), value.as_str());
346        }
347
348        if let Some(body) = body {
349            builder = match body {
350                RequestBody::Json(value) => builder.json(value)?,
351                RequestBody::JsonBytes(bytes) => builder.body_bytes(bytes.clone(), "application/json")?,
352            };
353        }
354
355        let request = builder.build()?;
356
357        // PUT请求使用专门的重试策略
358        let retry_context = if is_put_optimized {
359            format!("PUT_OPTIMIZED {}", path)
360        } else {
361            format!("{} {}", method, path)
362        };
363
364        // Use smart retry mechanism with PUT optimization
365        let retry_executor = if is_put_optimized {
366            &self.put_retry_executor // 使用PUT专用的快速重试器
367        } else {
368            &self.retry_executor // 使用通用重试器
369        };
370
371        retry_executor
372            .execute_with_context(&retry_context, || async {
373                // Execute with timeout
374                let result = tokio::time::timeout(timeout, async {
375                    let mut connection = if is_put_optimized && expected_size.unwrap_or(0) > 10240 {
376                        // 对于大的PUT请求，优先获取新连接
377                        self.get_fresh_connection().await?
378                    } else {
379                        self.get_connection().await?
380                    };
381
382                    match &mut connection {
383                        Either::Pool(conn) => {
384                            if let Some(stream) = conn.stream() {
385                                let result = send_request(stream, request.clone()).await;
386                                if result.is_err() {
387                                    conn.invalidate();
388                                }
389                                result
390                            } else {
391                                conn.invalidate();
392                                Err(KodeBridgeError::connection("Pooled connection is invalid"))
393                            }
394                        }
395                        Either::Direct(stream) => send_request(stream, request.clone()).await,
396                    }
397                })
398                .await;
399
400                match result {
401                    Ok(response) => response,
402                    Err(_) => Err(KodeBridgeError::timeout(timeout.as_millis() as u64)),
403                }
404            })
405            .await
406    }
407
408    /// Get a fresh connection optimized for PUT requests
409    async fn get_fresh_connection(&self) -> Result<Either<PooledConnection, IpcStream>> {
410        // 首先尝试从连接池获取新连接
411        if let Some(ref pool) = self.pool {
412            match tokio::time::timeout(Duration::from_millis(20), pool.get_fresh_connection()).await {
413                Ok(Ok(conn)) => return Ok(Either::Pool(conn)),
414                Ok(Err(_)) | Err(_) => {
415                    // 池化新连接失败，继续尝试直接连接
416                }
417            }
418        }
419
420        // 直接创建连接，使用更快的超时设置
421        match tokio::time::timeout(Duration::from_millis(100), IpcStream::connect(&self.endpoint)).await {
422            Ok(Ok(stream)) => Ok(Either::Direct(stream)),
423            Ok(Err(_)) | Err(_) => {
424                // 如果直接连接失败，回退到普通池化连接
425                if let Some(ref pool) = self.pool {
426                    let conn = pool.get_connection().await?;
427                    Ok(Either::Pool(conn))
428                } else {
429                    Err(KodeBridgeError::connection("Failed to get fresh connection"))
430                }
431            }
432        }
433    }
434
435    /// GET request
436    pub fn get(&self, path: &str) -> HttpRequestBuilder<'_> {
437        HttpRequestBuilder::new(self, Method::GET, path)
438    }
439
440    /// POST request
441    pub fn post(&self, path: &str) -> HttpRequestBuilder<'_> {
442        HttpRequestBuilder::new(self, Method::POST, path)
443    }
444
445    /// PUT request with optimization enabled by default
446    pub fn put(&self, path: &str) -> HttpRequestBuilder<'_> {
447        HttpRequestBuilder::new(self, Method::PUT, path)
448    }
449
450    /// Optimized batch PUT operations
451    pub async fn put_batch(&self, requests: Vec<(String, Value)>) -> Result<Vec<HttpResponse>> {
452        let batch_size = requests.len();
453        if batch_size == 0 {
454            return Ok(Vec::new());
455        }
456
457        let concurrent_limit = std::cmp::min(self.config.max_concurrent_requests, batch_size);
458        let mut responses = Vec::with_capacity(batch_size);
459        let mut pending = Vec::with_capacity(concurrent_limit);
460
461        for (path, body) in requests {
462            let body = Bytes::from(body.to_string());
463            pending.push(self.put(&path).json_bytes(body).optimize_for_put().send());
464
465            if pending.len() == concurrent_limit {
466                let chunk_results = futures::future::join_all(std::mem::take(&mut pending)).await;
467                for result in chunk_results {
468                    match result {
469                        Ok(response) => responses.push(response),
470                        Err(e) => return Err(e),
471                    }
472                }
473            }
474        }
475
476        if !pending.is_empty() {
477            let chunk_results = futures::future::join_all(pending).await;
478            for result in chunk_results {
479                match result {
480                    Ok(response) => responses.push(response),
481                    Err(e) => return Err(e),
482                }
483            }
484        }
485
486        Ok(responses)
487    }
488
489    /// DELETE request
490    pub fn delete(&self, path: &str) -> HttpRequestBuilder<'_> {
491        HttpRequestBuilder::new(self, Method::DELETE, path)
492    }
493
494    /// PATCH request
495    pub fn patch(&self, path: &str) -> HttpRequestBuilder<'_> {
496        HttpRequestBuilder::new(self, Method::PATCH, path)
497    }
498
499    /// HEAD request
500    pub fn head(&self, path: &str) -> HttpRequestBuilder<'_> {
501        HttpRequestBuilder::new(self, Method::HEAD, path)
502    }
503
504    /// OPTIONS request
505    pub fn options(&self, path: &str) -> HttpRequestBuilder<'_> {
506        HttpRequestBuilder::new(self, Method::OPTIONS, path)
507    }
508
509    /// Get pool statistics (if pooling is enabled)
510    pub fn pool_stats(&self) -> Option<crate::pool::PoolStats> {
511        self.pool.as_ref().map(|p| p.stats())
512    }
513
514    /// Close the client and clean up resources
515    pub fn close(&self) {
516        if let Some(ref pool) = self.pool {
517            pool.close();
518        }
519    }
520
521    /// Preheat connections for better PUT performance
522    pub async fn preheat_for_puts(&self, count: usize) {
523        if let Some(ref pool) = self.pool {
524            pool.preheat_for_puts(count).await;
525        }
526    }
527
528    /// Smart timeout calculation based on request characteristics
529    fn calculate_smart_timeout(&self, method: &str, body_size: Option<usize>) -> Duration {
530        match method {
531            "PUT" | "POST" => {
532                match body_size {
533                    Some(size) if size > 5 * 1024 * 1024 => Duration::from_secs(30), // >5MB: 30s
534                    Some(size) if size > 1024 * 1024 => Duration::from_secs(15),     // >1MB: 15s
535                    Some(size) if size > 100 * 1024 => Duration::from_secs(8),       // >100KB: 8s
536                    Some(size) if size > 10 * 1024 => Duration::from_secs(4),        // >10KB: 4s
537                    _ => Duration::from_secs(2),                                     // 小请求: 2s
538                }
539            }
540            _ => self.config.default_timeout, // 其他方法使用默认超时
541        }
542    }
543}
544
545impl<'a> HttpRequestBuilder<'a> {
546    fn new(client: &'a IpcHttpClient, method: Method, path: &str) -> Self {
547        let is_put = method == Method::PUT;
548        Self {
549            client,
550            method,
551            path: path.to_string(),
552            body: None,
553            timeout: None,
554            headers: Vec::new(),
555            put_optimized: is_put, // 自动为PUT请求启用优化
556            expected_size: None,
557        }
558    }
559
560    /// Set JSON body
561    pub fn json_body(mut self, body: &Value) -> Self {
562        self.body = Some(RequestBody::Json(body.clone()));
563        self
564    }
565
566    /// Set a pre-serialized JSON body.
567    pub fn json_bytes(mut self, body: Bytes) -> Self {
568        self.expected_size = Some(body.len());
569        self.body = Some(RequestBody::JsonBytes(body));
570        self
571    }
572
573    /// Set custom timeout
574    pub const fn timeout(mut self, timeout: Duration) -> Self {
575        self.timeout = Some(timeout);
576        self
577    }
578
579    /// 设置预期数据大小（用于PUT请求优化）
580    pub const fn expected_size(mut self, size: usize) -> Self {
581        self.expected_size = Some(size);
582        self
583    }
584
585    /// 启用PUT请求专门优化
586    pub const fn optimize_for_put(mut self) -> Self {
587        self.put_optimized = true;
588        self
589    }
590
591    /// Add custom header
592    pub fn header<K, V>(mut self, key: K, value: V) -> Self
593    where
594        K: Into<String>,
595        V: Into<String>,
596    {
597        self.headers.push((key.into(), value.into()));
598        self
599    }
600
601    /// Send the request
602    pub async fn send(self) -> Result<HttpResponse> {
603        let metrics = global_metrics();
604        let tracker = metrics.request_start(self.method.as_str());
605
606        // 为PUT请求优化超时设置，使用智能超时计算
607        let timeout = if self.put_optimized {
608            self.timeout.unwrap_or_else(|| {
609                self.client
610                    .calculate_smart_timeout(self.method.as_str(), self.expected_size)
611            })
612        } else {
613            self.timeout.unwrap_or(self.client.config.default_timeout)
614        };
615
616        match self
617            .client
618            .send_request_with_optimization(
619                self.method.as_str(),
620                &self.path,
621                self.body.as_ref(),
622                &self.headers,
623                timeout,
624                self.put_optimized,
625                self.expected_size,
626            )
627            .await
628        {
629            Ok(response) => {
630                tracker.success(response.status_code());
631                Ok(HttpResponse::new(response))
632            }
633            Err(e) => {
634                tracker.failure(&format!("{:?}", e));
635                Err(e)
636            }
637        }
638    }
639}
640
641/// Helper enum for connection types
642enum Either<A, B> {
643    Pool(A),
644    Direct(B),
645}
646
647impl Drop for IpcHttpClient {
648    fn drop(&mut self) {
649        self.close();
650    }
651}
kode_bridge/ipc_http_client.rs

kode_bridge/
ipc_http_client.rs
