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

objectiveai_sdk/http/
client.rs
