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

kode_bridge/
ipc_http_client.rs
