cratestack_axum/transport/
encode_sequence.rs

1use axum::http::{HeaderMap, StatusCode};
2use axum::response::Response;
3use cratestack_core::{CratestackError, RouteTransportCapabilities};
4use futures_util::{Stream, TryStreamExt};
5use serde::Serialize;
6
7use super::CBOR_SEQUENCE_CONTENT_TYPE;
8use super::http_transport::HttpTransport;
9use super::internal::fallback_error_response;
10use super::media_type::select_transport_response_content_type;
11
12pub fn encode_transport_sequence_result<TTransport, TValue>(
13    transport: &TTransport,
14    headers: &HeaderMap,
15    result: Result<Vec<TValue>, CratestackError>,
16) -> Response
17where
18    TTransport: HttpTransport,
19    TValue: Serialize,
20{
21    encode_transport_sequence_result_with_status_for(
22        transport,
23        headers,
24        &RouteTransportCapabilities {
25            request_types: &[],
26            response_types: &[],
27            default_response_type: "",
28            supports_sequence_response: false,
29        },
30        StatusCode::OK,
31        result,
32    )
33}
34
35pub fn encode_transport_sequence_result_with_status<TTransport, TValue>(
36    transport: &TTransport,
37    headers: &HeaderMap,
38    success_status: StatusCode,
39    result: Result<Vec<TValue>, CratestackError>,
40) -> Response
41where
42    TTransport: HttpTransport,
43    TValue: Serialize,
44{
45    encode_transport_sequence_result_with_status_for(
46        transport,
47        headers,
48        &RouteTransportCapabilities {
49            request_types: &[],
50            response_types: &[],
51            default_response_type: "",
52            supports_sequence_response: false,
53        },
54        success_status,
55        result,
56    )
57}
58
59pub fn encode_transport_sequence_result_with_status_for<TTransport, TValue>(
60    transport: &TTransport,
61    headers: &HeaderMap,
62    capabilities: &RouteTransportCapabilities,
63    success_status: StatusCode,
64    result: Result<Vec<TValue>, CratestackError>,
65) -> Response
66where
67    TTransport: HttpTransport,
68    TValue: Serialize,
69{
70    if !capabilities.supports_sequence_response {
71        return fallback_error_response(CratestackError::Internal(
72            "sequence response encoding requested for a route without sequence capability"
73                .to_owned(),
74        ));
75    }
76    let content_type =
77        match select_transport_response_content_type(transport, headers, capabilities) {
78            Ok(content_type) => content_type,
79            Err(error) => return fallback_error_response(error),
80        };
81    match result {
82        Ok(values) => transport
83            .encode_sequence_response(content_type, success_status, &values)
84            .unwrap_or_else(fallback_error_response),
85        Err(error) => {
86            let status = error.status_code();
87            let body = error.into_response();
88            transport
89                .encode_sequence_error_response(content_type, status, &body)
90                .unwrap_or_else(fallback_error_response)
91        }
92    }
93}
94
95/// Genuinely incremental counterpart to
96/// [`encode_transport_sequence_result_with_status_for`] for `@stream`
97/// procedures (cratestack#283): `result` carries the still-unconsumed
98/// item `Stream` rather than an already-collected `Vec`. `Err` here
99/// means a *preflight* failure (authorization, before anything was
100/// produced) — the ordinary buffered error path applies, since nothing
101/// has streamed to the client yet. A failure *during* the stream is a
102/// different thing entirely and never reaches this function as an
103/// `Err`: it's absorbed into the item stream itself as the tag-48900
104/// sentinel (see `super::stream_sequence`).
105///
106/// Only `application/cbor-seq` gets the truly incremental path. Any
107/// other negotiated content type (plain JSON/CBOR array) falls back to
108/// draining the stream into a `Vec` first and reusing the existing
109/// buffered encoder — arrays can't be flushed incrementally the same
110/// way, and `docs/design/rpc-transport.md` §3.3 only specifies
111/// incremental delivery for cbor-seq. This keeps a `@stream` procedure
112/// requested with a plain `Accept` behaving exactly like it did before
113/// this ticket.
114pub async fn encode_transport_stream_result_with_status_for<TTransport, TValue, S>(
115    transport: &TTransport,
116    headers: &HeaderMap,
117    capabilities: &RouteTransportCapabilities,
118    success_status: StatusCode,
119    result: Result<S, CratestackError>,
120) -> Response
121where
122    TTransport: HttpTransport,
123    TValue: Serialize + Send + 'static,
124    S: Stream<Item = Result<TValue, CratestackError>> + Send + 'static,
125{
126    if !capabilities.supports_sequence_response {
127        return fallback_error_response(CratestackError::Internal(
128            "sequence response encoding requested for a route without sequence capability"
129                .to_owned(),
130        ));
131    }
132    let content_type =
133        match select_transport_response_content_type(transport, headers, capabilities) {
134            Ok(content_type) => content_type,
135            Err(error) => return fallback_error_response(error),
136        };
137    match result {
138        Ok(stream) => {
139            if content_type == CBOR_SEQUENCE_CONTENT_TYPE {
140                transport
141                    .encode_sequence_stream_response(content_type, success_status, stream)
142                    .unwrap_or_else(fallback_error_response)
143            } else {
144                let values: Result<Vec<TValue>, CratestackError> = stream.try_collect().await;
145                encode_transport_sequence_result_with_status_for(
146                    transport,
147                    headers,
148                    capabilities,
149                    success_status,
150                    values,
151                )
152            }
153        }
154        Err(error) => {
155            let status = error.status_code();
156            let body = error.into_response();
157            transport
158                .encode_sequence_error_response(content_type, status, &body)
159                .unwrap_or_else(fallback_error_response)
160        }
161    }
162}
163
164#[cfg(test)]
165mod tests;
cratestack_axum/transport/encode_sequence.rs

cratestack_axum/transport/
encode_sequence.rs
