cratestack_client_rust/client/
streaming.rs

1use reqwest::Method;
2use serde::Serialize;
3use serde::de::DeserializeOwned;
4
5use crate::client::core::CratestackClient;
6use crate::client::decode::decode_sequence_response;
7use crate::codec::HttpClientCodec;
8use crate::error::{ClientError, HeaderPair};
9use crate::runtime::wire::{RuntimeRequestWire, RuntimeResponseWire};
10use crate::streaming::pump_streamed_response_typed;
11
12impl<C> CratestackClient<C>
13where
14    C: HttpClientCodec,
15{
16    pub async fn post_list<Input, Output>(
17        &self,
18        path: &str,
19        input: &Input,
20        headers: &[HeaderPair<'_>],
21    ) -> Result<Vec<Output>, ClientError>
22    where
23        Input: Serialize,
24        Output: DeserializeOwned,
25    {
26        if self.compact.is_some() {
27            super::compact_identity::refuse_duplicates(headers)?;
28        }
29        let body = self.codec.encode(input)?;
30        let response = self
31            .request_raw_with_query_and_accept(
32                Method::POST,
33                path,
34                Some(body),
35                None,
36                headers,
37                Some(self.codec.sequence_accept_header_value()),
38            )
39            .await?;
40        decode_sequence_response(&self.codec, &response)
41    }
42
43    /// Streaming variant of [`Self::post_list`]. Returns an
44    /// `mpsc::Receiver` that yields decoded items as they arrive over
45    /// the network — first-item latency drops from "buffer the whole
46    /// body" to "decode one chunk." Useful on mobile / flaky links
47    /// where time-to-first-byte matters more than total throughput.
48    ///
49    /// The receiver yields `Result<Output, ClientError>` per item.
50    /// Transport / decode errors are terminal — the next call to
51    /// `.recv()` returns `None` after one. A clean end-of-stream
52    /// (server closed cleanly after the last item) also surfaces as
53    /// `None` from the next `.recv()`.
54    ///
55    /// Ideally the server returns `application/cbor-seq`, decoded
56    /// incrementally as bytes arrive. If content negotiation instead
57    /// picks a buffered `application/cbor` or `application/json`
58    /// response, this still works — the body is buffered in full and
59    /// forwarded item-by-item through the same `Receiver` — but loses
60    /// the time-to-first-byte win, so prefer [`Self::post_list`] when
61    /// you already know the server won't stream. A `Content-Type` the
62    /// codec can't decode at all becomes a terminal `Err` on the
63    /// receiver naming what was received.
64    pub async fn post_list_streamed<Input, Output>(
65        &self,
66        path: &str,
67        input: &Input,
68        headers: &[HeaderPair<'_>],
69    ) -> Result<tokio::sync::mpsc::Receiver<Result<Output, ClientError>>, ClientError>
70    where
71        Input: Serialize,
72        Output: DeserializeOwned + Send + 'static,
73    {
74        if self.compact.is_some() {
75            super::compact_identity::refuse_duplicates(headers)?;
76        }
77        let body = self.codec.encode(input)?;
78        let (response, items) = self
79            .request_streamed_compact(
80                Method::POST,
81                path,
82                Some(body),
83                None,
84                headers,
85                self.codec.sequence_accept_header_value(),
86                true,
87            )
88            .await?;
89
90        // Bounded channel keeps memory tight on the consumer side —
91        // 16 items in flight is plenty for a single subscriber.
92        let (tx, rx) = tokio::sync::mpsc::channel(16);
93        crate::task::spawn(pump_streamed_response_typed::<C, Output, ClientError, _>(
94            self.codec.clone(),
95            response,
96            tx,
97            std::convert::identity,
98            items,
99        ));
100        Ok(rx)
101    }
102
103    pub async fn execute_raw_transport(
104        &self,
105        request: RuntimeRequestWire,
106    ) -> Result<RuntimeResponseWire, ClientError> {
107        let method = Method::from_bytes(request.method.as_bytes()).map_err(|error| {
108            ClientError::BadInput(format!("invalid HTTP method '{}': {error}", request.method))
109        })?;
110        let header_pairs = request
111            .headers
112            .iter()
113            .map(|header| (header.name.as_str(), header.value.as_str()))
114            .collect::<Vec<_>>();
115        let scoped = self.scoped_for_raw_path(&request.path)?;
116        scoped
117            .request_raw_with_query(
118                method,
119                &request.path,
120                if request.body.is_empty() {
121                    None
122                } else {
123                    Some(request.body)
124                },
125                request.canonical_query.as_deref(),
126                &header_pairs,
127            )
128            .await
129    }
130}
cratestack_client_rust/client/streaming.rs

cratestack_client_rust/client/
streaming.rs
