objectiveai_sdk/http/
client.rs

1//! HTTP client implementation for ObjectiveAI API.
2
3use crate::error;
4use eventsource_stream::Event as MessageEvent;
5use futures::{SinkExt, Stream, StreamExt};
6use reqwest_eventsource::{Event, RequestBuilderExt};
7use std::sync::Arc;
8use tokio_tungstenite::tungstenite;
9
10/// HTTP client for making requests to the ObjectiveAI API.
11///
12/// Handles authentication, request building, and response parsing for both
13/// unary and streaming endpoints.
14///
15/// # Example
16///
17/// ```ignore
18/// let client = HttpClient::new(
19///     reqwest::Client::new(),
20///     None, // Use default address
21///     Some("your-api-key"),
22///     None, // user_agent
23///     None, // x_title
24///     None, // http_referer
25///     None, // x_github_authorization
26///     None, // x_openrouter_authorization
27///     None, // x_mcp_authorization
28///     None, // agent_instance_hierarchy
29///     None, // mcp_call_timeout_ms
30/// );
31/// ```
32#[derive(Debug, Clone)]
33pub struct HttpClient {
34    /// The underlying reqwest HTTP client.
35    pub http_client: reqwest::Client,
36    /// Base URL for API requests. Defaults to `https://api.objectiveai.dev`.
37    pub address: String,
38    /// API key for authentication. Sent as `Bearer` token in `Authorization` header.
39    pub authorization: Option<Arc<String>>,
40    /// Value for the `User-Agent` header.
41    pub user_agent: Option<String>,
42    /// Value for the `X-Title` header.
43    pub x_title: Option<String>,
44    /// Value for both `Referer` and `HTTP-Referer` headers.
45    pub http_referer: Option<String>,
46    /// Value for the `X-GITHUB-AUTHORIZATION` header.
47    pub x_github_authorization: Option<Arc<String>>,
48    /// Value for the `X-OPENROUTER-AUTHORIZATION` header.
49    pub x_openrouter_authorization: Option<Arc<String>>,
50    /// Values for the `X-MCP-AUTHORIZATION` header (JSON-encoded).
51    pub x_mcp_authorization:
52        Option<Arc<std::collections::HashMap<String, String>>>,
53    /// Value for the `X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY` header.
54    pub agent_instance_hierarchy: Option<Arc<String>>,
55    /// Value for the `X-MCP-CALL-TIMEOUT` header, in integer
56    /// milliseconds: the per-request budget the API applies to each MCP
57    /// CALL its proxy makes on this request's behalf (HTTP and ws://
58    /// upstreams alike; never connects, never laboratory transfers).
59    /// Unset ⇒ no header ⇒ the API applies NO call timeout.
60    pub mcp_call_timeout_ms: Option<u64>,
61}
62
63impl HttpClient {
64    /// Creates a new HTTP client.
65    ///
66    /// # Arguments
67    ///
68    /// * `http_client` - The reqwest client to use for requests
69    /// * `address` - Base URL for API requests (defaults to `https://api.objectiveai.dev`)
70    /// * `authorization` - API key for authentication
71    /// * `user_agent` - Optional User-Agent header value
72    /// * `x_title` - Optional X-Title header value
73    /// * `http_referer` - Optional Referer header value
74    /// * `x_github_authorization` - Optional X-GITHUB-AUTHORIZATION header value
75    /// * `x_openrouter_authorization` - Optional X-OPENROUTER-AUTHORIZATION header value
76    /// * `x_mcp_authorization` - Optional X-MCP-AUTHORIZATION header value (HashMap)
77    /// * `agent_instance_hierarchy` - Optional X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY header value
78    /// * `mcp_call_timeout_ms` - Optional X-MCP-CALL-TIMEOUT header value
79    ///   (integer ms; no env fallback — config/option-only)
80    pub fn new(
81        http_client: reqwest::Client,
82        address: Option<impl Into<String>>,
83        authorization: Option<impl Into<String>>,
84        user_agent: Option<impl Into<String>>,
85        x_title: Option<impl Into<String>>,
86        http_referer: Option<impl Into<String>>,
87        x_github_authorization: Option<impl Into<String>>,
88        x_openrouter_authorization: Option<impl Into<String>>,
89        x_mcp_authorization: Option<std::collections::HashMap<String, String>>,
90        agent_instance_hierarchy: Option<impl Into<String>>,
91        mcp_call_timeout_ms: Option<u64>,
92    ) -> Self {
93        #[cfg(feature = "env")]
94        let env = |name: &str| -> Option<String> { std::env::var(name).ok() };
95
96        Self {
97            http_client,
98            address: match address {
99                Some(base) => base.into(),
100                #[cfg(feature = "env")]
101                None => env("OBJECTIVEAI_ADDRESS").unwrap_or_else(|| {
102                    "https://api.objectiveai.dev".to_string()
103                }),
104                #[cfg(not(feature = "env"))]
105                None => "https://api.objectiveai.dev".to_string(),
106            },
107            authorization: authorization.map(|k| Arc::new(k.into())).or_else(
108                || {
109                    #[cfg(feature = "env")]
110                    {
111                        env("OBJECTIVEAI_AUTHORIZATION").map(Arc::new)
112                    }
113                    #[cfg(not(feature = "env"))]
114                    {
115                        None
116                    }
117                },
118            ),
119            user_agent: user_agent.map(Into::into).or_else(|| {
120                #[cfg(feature = "env")]
121                {
122                    env("USER_AGENT")
123                }
124                #[cfg(not(feature = "env"))]
125                {
126                    None
127                }
128            }),
129            x_title: x_title.map(Into::into).or_else(|| {
130                #[cfg(feature = "env")]
131                {
132                    env("X_TITLE")
133                }
134                #[cfg(not(feature = "env"))]
135                {
136                    None
137                }
138            }),
139            http_referer: http_referer.map(Into::into).or_else(|| {
140                #[cfg(feature = "env")]
141                {
142                    env("HTTP_REFERER")
143                }
144                #[cfg(not(feature = "env"))]
145                {
146                    None
147                }
148            }),
149            x_github_authorization: x_github_authorization
150                .map(|v| Arc::new(v.into()))
151                .or_else(|| {
152                    #[cfg(feature = "env")]
153                    {
154                        env("GITHUB_AUTHORIZATION").map(Arc::new)
155                    }
156                    #[cfg(not(feature = "env"))]
157                    {
158                        None
159                    }
160                }),
161            x_openrouter_authorization: x_openrouter_authorization
162                .map(|v| Arc::new(v.into()))
163                .or_else(|| {
164                    #[cfg(feature = "env")]
165                    {
166                        env("OPENROUTER_AUTHORIZATION").map(Arc::new)
167                    }
168                    #[cfg(not(feature = "env"))]
169                    {
170                        None
171                    }
172                }),
173            x_mcp_authorization: x_mcp_authorization.map(Arc::new).or_else(
174                || {
175                    #[cfg(feature = "env")]
176                    {
177                        env("MCP_AUTHORIZATION")
178                            .and_then(|v| serde_json::from_str(&v).ok())
179                            .map(Arc::new)
180                    }
181                    #[cfg(not(feature = "env"))]
182                    {
183                        None
184                    }
185                },
186            ),
187            agent_instance_hierarchy: agent_instance_hierarchy.map(|v| Arc::new(v.into())).or_else(|| {
188                #[cfg(feature = "env")]
189                {
190                    env("OBJECTIVEAI_AGENT_INSTANCE_HIERARCHY").map(Arc::new)
191                }
192                #[cfg(not(feature = "env"))]
193                {
194                    None
195                }
196            }),
197            // Deliberately no env fallback: the caller (e.g. the daemon,
198            // from its `api.mcp_call_timeout_ms` config) supplies it or not.
199            mcp_call_timeout_ms,
200        }
201    }
202
203    /// Builds a request with authentication and custom headers.
204    fn request(
205        &self,
206        method: reqwest::Method,
207        path: &str,
208        body: Option<impl serde::Serialize>,
209    ) -> reqwest::RequestBuilder {
210        let url = format!(
211            "{}/{}",
212            self.address.trim_end_matches('/'),
213            path.trim_start_matches('/')
214        );
215        let mut request = self.http_client.request(method, &url);
216        if let Some(authorization) = &self.authorization {
217            let key = authorization
218                .strip_prefix("Bearer ")
219                .unwrap_or(authorization);
220            request =
221                request.header("authorization", format!("Bearer {}", key));
222        }
223        if let Some(user_agent) = &self.user_agent {
224            request = request.header("user-agent", user_agent);
225        }
226        if let Some(x_title) = &self.x_title {
227            request = request.header("x-title", x_title);
228        }
229        if let Some(http_referer) = &self.http_referer {
230            request = request.header("referer", http_referer);
231            request = request.header("http-referer", http_referer);
232        }
233        if let Some(token) = &self.x_github_authorization {
234            request = request.header("X-GITHUB-AUTHORIZATION", token.as_str());
235        }
236        if let Some(token) = &self.x_openrouter_authorization {
237            request =
238                request.header("X-OPENROUTER-AUTHORIZATION", token.as_str());
239        }
240        if let Some(headers) = &self.x_mcp_authorization {
241            if let Ok(json) = serde_json::to_string(headers.as_ref()) {
242                request = request.header("X-MCP-AUTHORIZATION", json);
243            }
244        }
245        if let Some(id) = &self.agent_instance_hierarchy {
246            request = request.header("X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY", id.as_str());
247        }
248        if let Some(ms) = self.mcp_call_timeout_ms {
249            request = request.header("X-MCP-CALL-TIMEOUT", ms.to_string());
250        }
251        if let Some(body) = body {
252            request = request.json(&body);
253        }
254        request
255    }
256
257    /// Sends a unary (request-response) API call and deserializes the response.
258    ///
259    /// # Type Parameters
260    ///
261    /// * `T` - The expected response type to deserialize into
262    ///
263    /// # Errors
264    ///
265    /// Returns [`super::HttpError`] if the request fails, returns a non-success status,
266    /// or the response cannot be deserialized.
267    pub async fn send_unary<T: serde::de::DeserializeOwned + Send + 'static>(
268        &self,
269        method: reqwest::Method,
270        path: impl AsRef<str>,
271        body: Option<impl serde::Serialize>,
272    ) -> Result<T, super::HttpError> {
273        let response = self
274            .http_client
275            .execute(
276                self.request(method, path.as_ref(), body)
277                    .build()
278                    .map_err(super::HttpError::RequestError)?,
279            )
280            .await
281            .map_err(super::HttpError::HttpError)?;
282        let code = response.status();
283        if code.is_success() {
284            let text =
285                response.text().await.map_err(super::HttpError::HttpError)?;
286            let mut de = serde_json::Deserializer::from_str(&text);
287            match serde_path_to_error::deserialize::<_, T>(&mut de) {
288                Ok(value) => Ok(value),
289                Err(e) => Err(super::HttpError::DeserializationError(e)),
290            }
291        } else {
292            match response.text().await {
293                Ok(text) => Err(super::HttpError::BadStatus {
294                    code,
295                    body: match serde_json::from_str::<serde_json::Value>(&text)
296                    {
297                        Ok(body) => body,
298                        Err(_) => serde_json::Value::String(text),
299                    },
300                }),
301                Err(_) => Err(super::HttpError::BadStatus {
302                    code,
303                    body: serde_json::Value::Null,
304                }),
305            }
306        }
307    }
308
309    /// Sends a unary API call that expects no response body.
310    ///
311    /// Useful for DELETE or other operations that only return a status code.
312    ///
313    /// # Errors
314    ///
315    /// Returns [`super::HttpError`] if the request fails or returns a non-success status.
316    pub async fn send_unary_no_response(
317        &self,
318        method: reqwest::Method,
319        path: impl AsRef<str>,
320        body: Option<impl serde::Serialize>,
321    ) -> Result<(), super::HttpError> {
322        let response = self
323            .http_client
324            .execute(
325                self.request(method, path.as_ref(), body)
326                    .build()
327                    .map_err(super::HttpError::RequestError)?,
328            )
329            .await
330            .map_err(super::HttpError::HttpError)?;
331        let code = response.status();
332        if code.is_success() {
333            Ok(())
334        } else {
335            match response.text().await {
336                Ok(text) => Err(super::HttpError::BadStatus {
337                    code,
338                    body: match serde_json::from_str::<serde_json::Value>(&text)
339                    {
340                        Ok(body) => body,
341                        Err(_) => serde_json::Value::String(text),
342                    },
343                }),
344                Err(_) => Err(super::HttpError::BadStatus {
345                    code,
346                    body: serde_json::Value::Null,
347                }),
348            }
349        }
350    }
351
352    /// Sends a streaming API call using Server-Sent Events (SSE).
353    ///
354    /// Returns a stream of deserialized chunks. The stream automatically handles:
355    /// - SSE `[DONE]` messages (filtered out)
356    /// - Comment lines starting with `:` (filtered out)
357    /// - Empty data lines (filtered out)
358    /// - API errors embedded in stream data
359    ///
360    /// # Type Parameters
361    ///
362    /// * `T` - The expected chunk type to deserialize each SSE message into
363    ///
364    /// # Errors
365    ///
366    /// Returns [`super::HttpError`] if the stream cannot be established.
367    pub async fn send_streaming<
368        T: serde::de::DeserializeOwned + Send + 'static,
369        P: AsRef<str> + Send,
370        B: serde::Serialize + Send,
371    >(
372        &self,
373        method: reqwest::Method,
374        path: P,
375        body: Option<B>,
376    ) -> Result<
377        impl Stream<Item = Result<T, super::HttpError>>
378        + Send
379        + 'static
380        + use<T, P, B>,
381        super::HttpError,
382    > {
383        // Stop the stream at [DONE] to prevent reqwest_eventsource from
384        // auto-reconnecting. Uses take_while on the raw SSE events, then
385        // maps/filters the remaining events into typed chunks.
386        // Stamps X-Transport: sse so the API's transport dispatcher
387        // routes this to the SSE branch (the API default is WS).
388        Ok(
389            self.request(method, path.as_ref(), body)
390                .header("X-Transport", "sse")
391                .eventsource()?
392                .take_while(|result| {
393                    let dominated = matches!(
394                        result,
395                        Ok(Event::Message(MessageEvent { data, .. })) if data == "[DONE]"
396                    );
397                    async move { !dominated }
398                })
399                .then(|result| async {
400                    match result {
401                        Ok(Event::Open) => None,
402                        Ok(Event::Message(MessageEvent { data, .. }))
403                            if data.starts_with(":")
404                                || data.is_empty() =>
405                        {
406                            None
407                        }
408                        Ok(Event::Message(MessageEvent { data, .. })) => {
409                            let mut de =
410                                serde_json::Deserializer::from_str(&data);
411                            Some(
412                                match serde_path_to_error::deserialize::<_, T>(
413                                    &mut de,
414                                ) {
415                                    Ok(value) => Ok(value),
416                                    Err(e) => match serde_json::from_str::<error::ResponseError>(&data) {
417                                        Ok(err) => Err(super::HttpError::ApiError(err)),
418                                        Err(_) => Err(super::HttpError::DeserializationError(e)),
419                                    },
420                                }
421                            )
422                        }
423                        Err(reqwest_eventsource::Error::InvalidStatusCode(
424                            code,
425                            response,
426                        )) => match response.text().await {
427                            Ok(body) => {
428                                Some(Err(super::HttpError::BadStatus {
429                                    code,
430                                    body: match serde_json::from_str::<
431                                        serde_json::Value,
432                                    >(
433                                        &body
434                                    ) {
435                                        Ok(body) => body,
436                                        Err(_) => {
437                                            serde_json::Value::String(body)
438                                        }
439                                    },
440                                }))
441                            }
442                            Err(_) => Some(Err(super::HttpError::BadStatus {
443                                code,
444                                body: serde_json::Value::Null,
445                            })),
446                        },
447                        Err(e) => Some(Err(super::HttpError::StreamError(e))),
448                    }
449                })
450                .filter_map(|x| async { x }),
451        )
452    }
453
454    /// WebSocket variant of [`Self::send_streaming`]. Opens a WS to
455    /// the configured `address`, sends `body` as the first text
456    /// frame, then demultiplexes inbound frames into:
457    ///
458    /// - Chunk frames (yielded on the returned [`Stream`]).
459    /// - [`client_response::Response`](crate::client_objectiveai_mcp::client_response::Response)
460    ///   frames (routed to the [`super::Notifier`]'s pending-id map).
461    /// - [`server_request::Request`](crate::client_objectiveai_mcp::server_request::Request)
462    ///   frames (dispatched to `handler`; the result is written back
463    ///   as a `server_response::Response` echoing the request id).
464    ///
465    /// Both the returned `Stream` and the returned [`super::Notifier`]
466    /// share the underlying WebSocket: dropping both stops the demux
467    /// task and closes the connection cleanly. Dropping only one
468    /// keeps the WS alive — useful when a caller wants to send
469    /// notifies after the chunk stream has finished, or vice-versa.
470    #[cfg(feature = "mcp")]
471    pub async fn send_streaming_ws<Chunk, B, H, P>(
472        &self,
473        method: reqwest::Method,
474        path: P,
475        body: B,
476        handler: H,
477    ) -> Result<
478        (
479            impl Stream<Item = Result<Chunk, super::HttpError>>
480            + Send
481            + Unpin
482            + 'static
483            + use<Chunk, B, H, P>,
484            super::Notifier,
485        ),
486        super::HttpError,
487    >
488    where
489        Chunk: serde::de::DeserializeOwned + Send + 'static,
490        B: serde::Serialize + Send + 'static,
491        H: super::McpHandler,
492        P: AsRef<str>,
493    {
494        use crate::client_objectiveai_mcp::{
495            client_response::Response as ClientResponse,
496            server_request::Request as ServerRequest,
497        };
498        use futures::stream::SplitStream;
499        use tokio::net::TcpStream;
500        use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
501
502        // Translate the configured `address` (http(s)://...) into a
503        // ws(s):// URL. Path is appended directly.
504        let url = format!(
505            "{}/{}",
506            self.address.trim_end_matches('/'),
507            path.as_ref().trim_start_matches('/')
508        );
509        let ws_url = if let Some(rest) = url.strip_prefix("https://") {
510            format!("wss://{rest}")
511        } else if let Some(rest) = url.strip_prefix("http://") {
512            format!("ws://{rest}")
513        } else {
514            url.clone()
515        };
516        let _ = method; // axum's WS route is `any(...)`, method is ignored on the wire.
517
518        // Build the upgrade request manually so we can apply the
519        // same auth + custom headers `request()` does for HTTP.
520        let mut req = tungstenite::handshake::client::Request::builder()
521            .method("GET")
522            .uri(&ws_url)
523            .header(
524                "Host",
525                reqwest::Url::parse(&url)
526                    .ok()
527                    .and_then(|u| u.host_str().map(str::to_owned))
528                    .unwrap_or_default(),
529            )
530            .header("Upgrade", "websocket")
531            .header("Connection", "Upgrade")
532            .header(
533                "Sec-WebSocket-Key",
534                tungstenite::handshake::client::generate_key(),
535            )
536            .header("Sec-WebSocket-Version", "13")
537            .header("X-Transport", "ws");
538        if let Some(authorization) = &self.authorization {
539            let key = authorization
540                .strip_prefix("Bearer ")
541                .unwrap_or(authorization.as_str());
542            req = req.header("authorization", format!("Bearer {}", key));
543        }
544        if let Some(ua) = &self.user_agent {
545            req = req.header("user-agent", ua);
546        }
547        if let Some(x_title) = &self.x_title {
548            req = req.header("x-title", x_title);
549        }
550        if let Some(http_referer) = &self.http_referer {
551            req = req.header("referer", http_referer);
552            req = req.header("http-referer", http_referer);
553        }
554        if let Some(token) = &self.x_github_authorization {
555            req = req.header("X-GITHUB-AUTHORIZATION", token.as_str());
556        }
557        if let Some(token) = &self.x_openrouter_authorization {
558            req = req.header("X-OPENROUTER-AUTHORIZATION", token.as_str());
559        }
560        if let Some(headers) = &self.x_mcp_authorization {
561            if let Ok(json) = serde_json::to_string(headers.as_ref()) {
562                req = req.header("X-MCP-AUTHORIZATION", json);
563            }
564        }
565        if let Some(id) = &self.agent_instance_hierarchy {
566            req = req.header("X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY", id.as_str());
567        }
568        if let Some(ms) = self.mcp_call_timeout_ms {
569            req = req.header("X-MCP-CALL-TIMEOUT", ms.to_string());
570        }
571        let req = req.body(()).map_err(|e| {
572            super::HttpError::WsConnect(tungstenite::Error::Http(
573                tungstenite::http::Response::builder()
574                    .status(400)
575                    .body(Some(e.to_string().into_bytes()))
576                    .unwrap(),
577            ))
578        })?;
579
580        let (ws_stream, _resp) = tokio_tungstenite::connect_async(req).await?;
581        let (mut sink, rx_stream): (
582            _,
583            SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
584        ) = ws_stream.split();
585
586        // Send the body as the first text frame.
587        let body_frame = serde_json::to_string(&body)
588            .map_err(super::HttpError::NotifySerialize)?;
589        sink.send(tungstenite::Message::Text(body_frame.into()))
590            .await
591            .map_err(super::HttpError::NotifySend)?;
592
593        // Build the per-connection state shared with Notifier + demux.
594        let sink: super::notifier::SharedSink =
595            Arc::new(tokio::sync::Mutex::new(sink));
596        let pending: super::notifier::PendingNotifies =
597            Arc::new(dashmap::DashMap::new());
598
599        // mpsc the demux task pushes chunks (or terminal errors) into.
600        // Use futures::channel::mpsc so the rx side is `impl Stream`
601        // without pulling in tokio-stream.
602        let (chunk_tx, chunk_rx) = futures::channel::mpsc::unbounded::<
603            Result<Chunk, super::HttpError>,
604        >();
605
606        let demux_sink = sink.clone();
607        let demux_pending = pending.clone();
608        let handler = Arc::new(handler);
609        tokio::spawn(async move {
610            let mut rx_stream = rx_stream;
611            let mut chunk_tx = chunk_tx;
612            loop {
613                let msg = match rx_stream.next().await {
614                    Some(m) => m,
615                    None => break,
616                };
617                let text = match msg {
618                    Ok(tungstenite::Message::Text(t)) => {
619                        let s = t.to_string();
620                        s
621                    }
622                    Ok(tungstenite::Message::Binary(_)) => {
623                        continue;
624                    }
625                    Ok(
626                        tungstenite::Message::Ping(_)
627                        | tungstenite::Message::Pong(_),
628                    ) => continue,
629                    Ok(tungstenite::Message::Close(_)) => {
630                        break;
631                    }
632                    Ok(tungstenite::Message::Frame(_)) => continue,
633                    Err(_) => {
634                        break;
635                    }
636                };
637
638                // Classification: try client_response, then
639                // server_request, then chunk. Order matters — chunks
640                // tend to have many fields; the envelopes have a
641                // distinctive `id` + tagged `type`.
642                if let Ok(response) =
643                    serde_json::from_str::<ClientResponse>(&text)
644                {
645                    let id = response.id().to_string();
646                    if let Some((_, tx)) = demux_pending.remove(&id) {
647                        let _ = tx.send(response);
648                    }
649                    continue;
650                }
651                if let Ok(request) =
652                    serde_json::from_str::<ServerRequest>(&text)
653                {
654                    let id = request.id.clone();
655                    let handler = handler.clone();
656                    let demux_sink = demux_sink.clone();
657                    tokio::spawn(async move {
658                        let id = id;
659                        // Handler returns the full response (incl.
660                        // matching id); we just frame + write it.
661                        let response = handler.handle(request).await;
662                        let frame = match serde_json::to_string(&response) {
663                            Ok(s) => s,
664                            Err(_) => {
665                                return;
666                            }
667                        };
668                        let mut guard = demux_sink.lock().await;
669                        let send_result = guard
670                            .send(tungstenite::Message::Text(frame.into()))
671                            .await;
672                    });
673                    continue;
674                }
675
676                // Chunk path.
677                let mut de = serde_json::Deserializer::from_str(&text);
678                match serde_path_to_error::deserialize::<_, Chunk>(&mut de) {
679                    Ok(chunk) => {
680                        if chunk_tx.unbounded_send(Ok(chunk)).is_err() {
681                            break;
682                        }
683                    }
684                    Err(e) => {
685                        // Try to parse as a ResponseError before
686                        // surfacing the deserialization failure.
687                        let err = match serde_json::from_str::<
688                            error::ResponseError,
689                        >(&text)
690                        {
691                            Ok(api_err) => super::HttpError::ApiError(api_err),
692                            Err(_) => super::HttpError::DeserializationError(e),
693                        };
694                        let _ = chunk_tx.unbounded_send(Err(err));
695                        break;
696                    }
697                }
698            }
699            // Make sure any awaiting Notifier futures unblock when
700            // we exit — dropping the dashmap fires every oneshot
701            // Sender's drop, which causes the rx side to error.
702            drop(demux_pending);
703            drop(chunk_tx);
704        });
705
706        let notifier = super::Notifier::new(sink, pending);
707        Ok((chunk_rx, notifier))
708    }
709}
objectiveai_sdk/http/client.rs

objectiveai_sdk/http/
client.rs
