kode_bridge/
ipc_stream_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::RequestBuilder;
9use crate::stream_client::{send_streaming_request, StreamingResponse};
10use crate::transport::{Endpoint, IpcStream};
11use http::Method;
12use std::str::FromStr as _;
13use tracing::{debug, trace};
14
15/// Configuration for IPC streaming client
16#[derive(Debug, Clone)]
17pub struct StreamClientConfig {
18    /// Default timeout for connections
19    pub default_timeout: Duration,
20    /// Maximum number of retry attempts
21    pub max_retries: usize,
22    /// Delay between retry attempts
23    pub retry_delay: Duration,
24    /// Buffer size for streaming
25    pub buffer_size: usize,
26}
27
28impl Default for StreamClientConfig {
29    fn default() -> Self {
30        Self {
31            default_timeout: Duration::from_secs(60),
32            max_retries: 3,
33            retry_delay: Duration::from_millis(100),
34            buffer_size: 8192,
35        }
36    }
37}
38
39/// Specialized IPC streaming client for handling real-time data streams
40pub struct IpcStreamClient {
41    endpoint: Endpoint,
42    config: StreamClientConfig,
43}
44
45/// Stream request builder for fluent API
46pub struct StreamRequestBuilder<'a> {
47    client: &'a IpcStreamClient,
48    method: Method,
49    path: String,
50    body: Option<Value>,
51    timeout: Duration,
52    headers: Vec<(String, String)>,
53}
54
55/// Stream response wrapper with chainable methods
56pub struct StreamResponse {
57    inner: StreamingResponse,
58}
59
60impl StreamResponse {
61    const fn new(response: StreamingResponse) -> Self {
62        Self { inner: response }
63    }
64
65    /// Get the HTTP status code
66    pub const fn status(&self) -> u16 {
67        self.inner.status_code()
68    }
69
70    /// Check if response indicates success (2xx status)
71    pub fn is_success(&self) -> bool {
72        self.inner.is_success()
73    }
74
75    /// Check if response indicates client error (4xx status)
76    pub fn is_client_error(&self) -> bool {
77        self.inner.is_client_error()
78    }
79
80    /// Check if response indicates server error (5xx status)
81    pub fn is_server_error(&self) -> bool {
82        self.inner.is_server_error()
83    }
84
85    /// Convert stream to JSON objects with automatic parsing and timeout
86    pub async fn json_results<T>(self) -> Result<Vec<T>>
87    where
88        T: DeserializeOwned + Send,
89    {
90        self.inner.json(Duration::from_secs(30)).await
91    }
92
93    /// Convert stream to JSON objects with custom timeout
94    pub async fn json<T>(self, timeout: Duration) -> Result<Vec<T>>
95    where
96        T: DeserializeOwned + Send,
97    {
98        self.inner.json(timeout).await
99    }
100
101    /// Process stream in real-time with a handler
102    pub async fn process_lines<F>(self, timeout: Duration, mut handler: F) -> Result<()>
103    where
104        F: FnMut(&str) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
105    {
106        self.inner
107            .process_lines_with_timeout(timeout, |line| {
108                handler(line).map(|_| true) // Convert to continue flag
109            })
110            .await
111    }
112
113    /// Process stream with custom JSON processing
114    pub async fn process_json<F, T>(self, timeout: Duration, handler: F) -> Result<Vec<T>>
115    where
116        F: FnMut(&str) -> Option<T> + Send,
117        T: Send + 'static,
118    {
119        self.inner.process_json(timeout, handler).await
120    }
121
122    /// Collect all stream content as text
123    pub async fn collect_text(self) -> Result<String> {
124        self.inner.collect_text().await
125    }
126
127    /// Collect stream content with timeout
128    pub async fn collect_text_with_timeout(self, timeout: Duration) -> Result<String> {
129        self.inner.collect_text_with_timeout(timeout).await
130    }
131
132    /// Get the underlying modern StreamingResponse
133    pub fn into_inner(self) -> StreamingResponse {
134        self.inner
135    }
136
137    /// Get headers as JSON for backward compatibility
138    pub fn headers(&self) -> Value {
139        self.inner.headers_json()
140    }
141}
142
143impl IpcStreamClient {
144    /// Create a new IPC streaming client with default configuration
145    pub fn new<P>(path: P) -> Result<Self>
146    where
147        P: AsRef<Path>,
148    {
149        Self::with_config(path, StreamClientConfig::default())
150    }
151
152    /// Create a new IPC streaming client with custom configuration
153    pub fn with_config<P>(path: P, config: StreamClientConfig) -> Result<Self>
154    where
155        P: AsRef<Path>,
156    {
157        let endpoint = Endpoint::new(path)?;
158
159        Ok(Self { endpoint, config })
160    }
161
162    /// Create a connection with retry logic
163    async fn create_connection(&self) -> Result<IpcStream> {
164        let mut last_error = None;
165
166        for attempt in 0..self.config.max_retries {
167            if attempt > 0 {
168                tokio::time::sleep(self.config.retry_delay).await;
169            }
170
171            match IpcStream::connect(&self.endpoint).await {
172                Ok(stream) => {
173                    debug!("Created streaming connection on attempt {}", attempt + 1);
174                    return Ok(stream);
175                }
176                Err(e) => {
177                    trace!("Streaming connection attempt {} failed: {}", attempt + 1, e);
178                    last_error = Some(e);
179                }
180            }
181        }
182
183        Err(KodeBridgeError::connection(format!(
184            "Failed to create streaming connection after {} attempts: {}",
185            self.config.max_retries,
186            last_error
187                .map(|e| e.to_string())
188                .unwrap_or_else(|| "Unknown error".to_string())
189        )))
190    }
191
192    /// Internal method to send streaming requests
193    async fn send_request_internal(
194        &self,
195        method: &str,
196        path: &str,
197        body: Option<&Value>,
198        headers: &[(String, String)],
199        timeout: Duration,
200    ) -> Result<StreamingResponse> {
201        let method =
202            Method::from_str(method).map_err(|e| KodeBridgeError::invalid_request(format!("Invalid method: {}", e)))?;
203
204        let mut builder = RequestBuilder::new(method, path.to_string());
205
206        for (key, value) in headers {
207            builder = builder.header(key, value);
208        }
209
210        if let Some(json_body) = body {
211            builder = builder.json(json_body)?;
212        }
213
214        let request = builder.build()?;
215
216        // Execute with timeout
217        let result = tokio::time::timeout(timeout, async {
218            let stream = self.create_connection().await?;
219            send_streaming_request(stream, request).await
220        })
221        .await;
222
223        match result {
224            Ok(response) => response,
225            Err(_) => Err(KodeBridgeError::timeout(timeout.as_millis() as u64)),
226        }
227    }
228
229    /// GET streaming request
230    pub fn get(&self, path: &str) -> StreamRequestBuilder<'_> {
231        StreamRequestBuilder::new(self, Method::GET, path)
232    }
233
234    /// POST streaming request
235    pub fn post(&self, path: &str) -> StreamRequestBuilder<'_> {
236        StreamRequestBuilder::new(self, Method::POST, path)
237    }
238
239    /// PUT streaming request
240    pub fn put(&self, path: &str) -> StreamRequestBuilder<'_> {
241        StreamRequestBuilder::new(self, Method::PUT, path)
242    }
243
244    /// DELETE streaming request
245    pub fn delete(&self, path: &str) -> StreamRequestBuilder<'_> {
246        StreamRequestBuilder::new(self, Method::DELETE, path)
247    }
248
249    /// PATCH streaming request
250    pub fn patch(&self, path: &str) -> StreamRequestBuilder<'_> {
251        StreamRequestBuilder::new(self, Method::PATCH, path)
252    }
253
254    /// HEAD streaming request
255    pub fn head(&self, path: &str) -> StreamRequestBuilder<'_> {
256        StreamRequestBuilder::new(self, Method::HEAD, path)
257    }
258
259    /// OPTIONS streaming request
260    pub fn options(&self, path: &str) -> StreamRequestBuilder<'_> {
261        StreamRequestBuilder::new(self, Method::OPTIONS, path)
262    }
263}
264
265impl<'a> StreamRequestBuilder<'a> {
266    fn new(client: &'a IpcStreamClient, method: Method, path: &str) -> Self {
267        Self {
268            client,
269            method,
270            path: path.to_string(),
271            body: None,
272            timeout: client.config.default_timeout,
273            headers: Vec::new(),
274        }
275    }
276
277    /// Set JSON body
278    pub fn json_body(mut self, body: &Value) -> Self {
279        self.body = Some(body.clone());
280        self
281    }
282
283    /// Set custom timeout
284    pub const fn timeout(mut self, timeout: Duration) -> Self {
285        self.timeout = timeout;
286        self
287    }
288
289    /// Add custom header
290    pub fn header<K, V>(mut self, key: K, value: V) -> Self
291    where
292        K: Into<String>,
293        V: Into<String>,
294    {
295        self.headers.push((key.into(), value.into()));
296        self
297    }
298
299    /// Send the streaming request
300    pub async fn send(self) -> Result<StreamResponse> {
301        let response = self
302            .client
303            .send_request_internal(
304                self.method.as_str(),
305                &self.path,
306                self.body.as_ref(),
307                &self.headers,
308                self.timeout,
309            )
310            .await?;
311
312        Ok(StreamResponse::new(response))
313    }
314
315    /// Send and get JSON results automatically (convenience method)
316    pub async fn json_results<T>(self) -> Result<Vec<T>>
317    where
318        T: DeserializeOwned + Send,
319    {
320        let response = self.send().await?;
321        response.json_results().await
322    }
323
324    /// Send and process lines with a handler (convenience method)
325    pub async fn process_lines<F>(self, handler: F) -> Result<()>
326    where
327        F: FnMut(&str) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
328    {
329        let timeout = self.timeout;
330        let response = self.send().await?;
331        response.process_lines(timeout, handler).await
332    }
333}
kode_bridge/ipc_stream_client.rs

kode_bridge/
ipc_stream_client.rs
