gproxy_protocol/adapt/
json.rs

1use crate::{
2    HttpBody, WireRequest, WireResponse,
3    capability::Upstream,
4    codec::{self, CodecError, CodecErrorKind, CodecLimits},
5    transform::{TransformError, TransformErrorKind},
6    wire::DeclaredFields,
7};
8use serde::{Serialize, de::DeserializeOwned};
9
10/// HTTP errors retain their original body and metadata for the operation's
11/// concrete error converter. A non-2xx response is never a transport error.
12#[derive(Debug)]
13pub enum JsonInvocation<T> {
14    Success(WireResponse<T>),
15    Rejected(WireResponse<HttpBody>),
16}
17
18/// Send exactly one already-prepared, typed target request. This helper neither
19/// routes nor retries. Dropping its future cancels further adapter work; any
20/// upstream side effect remains subject to that operation's recovery contract.
21///
22/// Only declared fields are encoded/returned. JSON Schema, arguments and other
23/// formal JSON values remain intact. The host enforces deadlines and stream idle
24/// limits, and replaces authentication according to [`Upstream::send`].
25pub async fn invoke_json<U, I, O>(
26    upstream: &U,
27    target: &U::Target,
28    request: WireRequest<I>,
29    limits: CodecLimits,
30) -> Result<JsonInvocation<O>, TransformError>
31where
32    U: Upstream,
33    I: Serialize + DeclaredFields,
34    O: DeserializeOwned + DeclaredFields,
35{
36    let WireRequest {
37        method,
38        path,
39        query,
40        mut headers,
41        body,
42    } = request;
43
44    let host_limits = upstream.limits();
45    let write_limits = bound_limits(limits, host_limits.write_bytes);
46    let body = codec::encode_json(&body.into_declared(), write_limits)
47        .map_err(|error| codec_error(error, false))?;
48    // The target JSON bytes replace any former representation of this request.
49    for name in [
50        http::header::CONTENT_LENGTH,
51        http::header::CONTENT_ENCODING,
52        http::header::TRANSFER_ENCODING,
53    ] {
54        headers.remove(name);
55    }
56    headers.insert(
57        http::header::CONTENT_TYPE,
58        http::HeaderValue::from_static("application/json"),
59    );
60    let response = upstream
61        .send(
62            target,
63            WireRequest {
64                method,
65                path,
66                query,
67                headers,
68                body: HttpBody::Bytes(body),
69            },
70        )
71        .await?;
72    receive_json(response, limits, host_limits.read_bytes).await
73}
74
75/// Send one request with an empty HTTP body and decode its JSON success body.
76/// This is suitable for models/files GET operations; `()` is not encoded as
77/// JSON `null`. Non-2xx bodies are returned untouched, as with [`invoke_json`].
78pub async fn invoke_empty<U, O>(
79    upstream: &U,
80    target: &U::Target,
81    request: WireRequest<()>,
82    limits: CodecLimits,
83) -> Result<JsonInvocation<O>, TransformError>
84where
85    U: Upstream,
86    O: DeserializeOwned + DeclaredFields,
87{
88    let WireRequest {
89        method,
90        path,
91        query,
92        mut headers,
93        body: (),
94    } = request;
95    for name in [
96        http::header::CONTENT_TYPE,
97        http::header::CONTENT_LENGTH,
98        http::header::CONTENT_ENCODING,
99        http::header::TRANSFER_ENCODING,
100    ] {
101        headers.remove(name);
102    }
103    let response = upstream
104        .send(
105            target,
106            WireRequest {
107                method,
108                path,
109                query,
110                headers,
111                body: HttpBody::Bytes(bytes::Bytes::new()),
112            },
113        )
114        .await?;
115    receive_json(response, limits, upstream.limits().read_bytes).await
116}
117
118async fn receive_json<O: DeserializeOwned + DeclaredFields>(
119    response: WireResponse<HttpBody>,
120    limits: CodecLimits,
121    read_bytes: u64,
122) -> Result<JsonInvocation<O>, TransformError> {
123    if !response.status.is_success() {
124        return Ok(JsonInvocation::Rejected(response));
125    }
126    let WireResponse {
127        status,
128        mut headers,
129        body,
130    } = response;
131    let limits = bound_limits(limits, read_bytes);
132    let bytes = codec::read_http_body(body, limits)
133        .await
134        .map_err(|error| codec_error(error, true))?;
135    let body: O = codec::decode_json(&bytes, limits).map_err(|error| codec_error(error, true))?;
136    // The returned typed body has been rebuilt, so upstream representation
137    // lengths/encodings cannot describe a later serialization of that value.
138    for name in [
139        http::header::CONTENT_LENGTH,
140        http::header::CONTENT_ENCODING,
141        http::header::TRANSFER_ENCODING,
142    ] {
143        headers.remove(name);
144    }
145    headers.insert(
146        http::header::CONTENT_TYPE,
147        http::HeaderValue::from_static("application/json"),
148    );
149    Ok(JsonInvocation::Success(WireResponse {
150        status,
151        headers,
152        body: body.into_declared(),
153    }))
154}
155
156fn bound_limits(mut limits: CodecLimits, host_bytes: u64) -> CodecLimits {
157    limits.max_body_bytes = limits.max_body_bytes.min(host_bytes);
158    limits.max_buffer_bytes = limits.max_buffer_bytes.min(host_bytes);
159    limits.max_value_bytes = limits.max_value_bytes.min(host_bytes);
160    limits
161}
162
163fn codec_error(error: CodecError, response: bool) -> TransformError {
164    let kind = match error.kind() {
165        CodecErrorKind::Limit => TransformErrorKind::Limit,
166        CodecErrorKind::Transport => TransformErrorKind::Host,
167        CodecErrorKind::Invalid
168        | CodecErrorKind::UnexpectedEof
169        | CodecErrorKind::Utf8
170        | CodecErrorKind::Json
171        | CodecErrorKind::Multipart => {
172            if response {
173                TransformErrorKind::InvalidResult
174            } else {
175                TransformErrorKind::InvalidInput
176            }
177        }
178    };
179    let detail = error.to_string();
180    TransformError::with_source(
181        kind,
182        if response {
183            "upstream.response"
184        } else {
185            "upstream.request"
186        },
187        detail,
188        error,
189    )
190}
gproxy_protocol/adapt/json.rs

gproxy_protocol/adapt/
json.rs
