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 (error, tag) = crate::idempotency::answered(error);
87            let status = error.status_code();
88            let challenges = crate::auth_challenge::challenges_of(&error);
89            let body = error.into_response();
90            let response = transport
91                .encode_sequence_error_response(content_type, status, &body)
92                .unwrap_or_else(fallback_error_response);
93            let response = crate::auth_challenge::attach(response, &challenges);
94            crate::idempotency::with_tag(tag, response)
95        }
96    }
97}
98
99/// Genuinely incremental counterpart to
100/// [`encode_transport_sequence_result_with_status_for`] for `@stream`
101/// procedures (cratestack#283): `result` carries the still-unconsumed
102/// item `Stream` rather than an already-collected `Vec`. `Err` here
103/// means a *preflight* failure (authorization, before anything was
104/// produced) — the ordinary buffered error path applies, since nothing
105/// has streamed to the client yet. A failure *during* the stream is a
106/// different thing entirely and never reaches this function as an
107/// `Err`: it's absorbed into the item stream itself as the tag-48900
108/// sentinel (see `super::stream_sequence`).
109///
110/// Only `application/cbor-seq` gets the truly incremental path. Any
111/// other negotiated content type (plain JSON/CBOR array) falls back to
112/// draining the stream into a `Vec` first and reusing the existing
113/// buffered encoder — arrays can't be flushed incrementally the same
114/// way, and `docs/design/rpc-transport.md` §3.3 only specifies
115/// incremental delivery for cbor-seq. This keeps a `@stream` procedure
116/// requested with a plain `Accept` behaving exactly like it did before
117/// this ticket.
118pub async fn encode_transport_stream_result_with_status_for<TTransport, TValue, S>(
119    transport: &TTransport,
120    headers: &HeaderMap,
121    capabilities: &RouteTransportCapabilities,
122    success_status: StatusCode,
123    result: Result<S, CratestackError>,
124) -> Response
125where
126    TTransport: HttpTransport,
127    TValue: Serialize + Send + 'static,
128    S: Stream<Item = Result<TValue, CratestackError>> + Send + 'static,
129{
130    if !capabilities.supports_sequence_response {
131        return fallback_error_response(CratestackError::Internal(
132            "sequence response encoding requested for a route without sequence capability"
133                .to_owned(),
134        ));
135    }
136    let content_type =
137        match select_transport_response_content_type(transport, headers, capabilities) {
138            Ok(content_type) => content_type,
139            Err(error) => return fallback_error_response(error),
140        };
141    match result {
142        Ok(stream) => {
143            if content_type == CBOR_SEQUENCE_CONTENT_TYPE {
144                transport
145                    .encode_sequence_stream_response(content_type, success_status, stream)
146                    .unwrap_or_else(fallback_error_response)
147            } else {
148                let values: Result<Vec<TValue>, CratestackError> = stream.try_collect().await;
149                encode_transport_sequence_result_with_status_for(
150                    transport,
151                    headers,
152                    capabilities,
153                    success_status,
154                    values,
155                )
156            }
157        }
158        Err(error) => {
159            let (error, tag) = crate::idempotency::answered(error);
160            let status = error.status_code();
161            let challenges = crate::auth_challenge::challenges_of(&error);
162            let body = error.into_response();
163            let response = transport
164                .encode_sequence_error_response(content_type, status, &body)
165                .unwrap_or_else(fallback_error_response);
166            let response = crate::auth_challenge::attach(response, &challenges);
167            crate::idempotency::with_tag(tag, response)
168        }
169    }
170}
171
172#[cfg(test)]
173mod tests;
cratestack_axum/transport/encode_sequence.rs

cratestack_axum/transport/
encode_sequence.rs
