connect_rpc/
response.rs

1pub mod builder;
2pub mod error;
3
4use http::{header, HeaderMap, StatusCode};
5
6use crate::{
7    common::{
8        streaming_message_codec, unary_message_codec, CONNECT_CONTENT_ENCODING,
9        CONTENT_ENCODING_IDENTITY,
10    },
11    metadata::Metadata,
12    request::ConnectRequest,
13    Error,
14};
15
16/// A Connect response.
17pub trait ConnectResponse {
18    /// Returns the status code.
19    fn status(&self) -> StatusCode;
20
21    /// Returns the message codec.
22    fn message_codec(&self) -> Result<&str, Error>;
23
24    /// Returns the content encoding.
25    fn content_encoding(&self) -> Option<&str>;
26
27    /// Returns a reference to the metadata.
28    fn metadata(&self) -> &impl Metadata;
29
30    /// Validates the response.
31    fn validate(&self, opts: &ValidateOpts) -> Result<(), Error>;
32}
33
34/// Options for [`ConnectResponse::validate`].
35#[derive(Clone, Debug, Default)]
36pub struct ValidateOpts {
37    /// If given, the response message codec must match.
38    pub message_codec: Option<String>,
39    /// If given, the response content encoding must match (or be 'identity').
40    pub accept_encoding: Option<Vec<String>>,
41}
42
43impl ValidateOpts {
44    pub fn from_request(req: &impl ConnectRequest) -> Self {
45        let message_codec = req.message_codec().map(ToString::to_string).ok();
46        let accept_encoding = Some(req.accept_encoding().map(ToString::to_string).collect());
47        Self {
48            message_codec,
49            accept_encoding,
50        }
51    }
52}
53
54trait HttpConnectResponse {
55    fn http_status(&self) -> StatusCode;
56
57    fn http_headers(&self) -> &HeaderMap;
58
59    fn http_message_codec(&self) -> Result<&str, Error>;
60
61    fn http_content_encoding(&self) -> Option<&str>;
62}
63
64impl<T: HttpConnectResponse> ConnectResponse for T {
65    fn status(&self) -> StatusCode {
66        self.http_status()
67    }
68
69    fn message_codec(&self) -> Result<&str, Error> {
70        self.http_message_codec()
71    }
72
73    fn content_encoding(&self) -> Option<&str> {
74        self.http_content_encoding()
75    }
76
77    fn metadata(&self) -> &impl Metadata {
78        self.http_headers()
79    }
80
81    fn validate(&self, opts: &ValidateOpts) -> Result<(), Error> {
82        let codec = self.message_codec()?;
83        if let Some(validate_codec) = &opts.message_codec {
84            if codec != validate_codec {
85                return Err(Error::UnexpectedMessageCodec(codec.into()));
86            }
87        }
88        if let Some(encoding) = self.content_encoding() {
89            if encoding != CONTENT_ENCODING_IDENTITY {
90                if let Some(accept_encoding) = &opts.accept_encoding {
91                    if !accept_encoding.iter().any(|accept| accept == encoding) {
92                        return Err(Error::UnacceptableEncoding(encoding.into()));
93                    }
94                }
95            }
96        }
97        Ok(())
98    }
99}
100
101#[derive(Clone, Debug)]
102pub struct UnaryResponse<T>(http::Response<T>);
103
104impl<T> UnaryResponse<T> {
105    pub fn body(&self) -> &T {
106        self.0.body()
107    }
108}
109
110impl<T: AsRef<[u8]>> UnaryResponse<T> {
111    pub fn result(self, validate_opts: &ValidateOpts) -> Result<Self, Error> {
112        if !self.0.status().is_success() {
113            return Err(Error::ConnectError(http::Response::from(self).into()));
114        }
115        self.validate(validate_opts)?;
116        Ok(self)
117    }
118}
119
120impl<T> HttpConnectResponse for UnaryResponse<T> {
121    fn http_status(&self) -> StatusCode {
122        self.0.status()
123    }
124
125    fn http_headers(&self) -> &HeaderMap {
126        self.0.headers()
127    }
128
129    fn http_message_codec(&self) -> Result<&str, Error> {
130        unary_message_codec(self.http_headers())
131    }
132
133    fn http_content_encoding(&self) -> Option<&str> {
134        self.http_headers()
135            .get(header::CONTENT_ENCODING)?
136            .to_str()
137            .ok()
138    }
139}
140
141impl<T> From<http::Response<T>> for UnaryResponse<T> {
142    fn from(resp: http::Response<T>) -> Self {
143        Self(resp)
144    }
145}
146
147impl<T> From<UnaryResponse<T>> for http::Response<T> {
148    fn from(resp: UnaryResponse<T>) -> Self {
149        resp.0
150    }
151}
152
153#[derive(Clone, Debug)]
154pub struct StreamingResponse<T>(http::Response<T>);
155
156impl<T> HttpConnectResponse for StreamingResponse<T> {
157    fn http_status(&self) -> StatusCode {
158        self.0.status()
159    }
160
161    fn http_headers(&self) -> &HeaderMap {
162        self.0.headers()
163    }
164
165    fn http_message_codec(&self) -> Result<&str, Error> {
166        streaming_message_codec(self.http_headers())
167    }
168
169    fn http_content_encoding(&self) -> Option<&str> {
170        self.http_headers()
171            .get(CONNECT_CONTENT_ENCODING)?
172            .to_str()
173            .ok()
174    }
175}
176
177impl<T> From<http::Response<T>> for StreamingResponse<T> {
178    fn from(resp: http::Response<T>) -> Self {
179        Self(resp)
180    }
181}
182
183impl<T> From<StreamingResponse<T>> for http::Response<T> {
184    fn from(resp: StreamingResponse<T>) -> Self {
185        resp.0
186    }
187}
connect_rpc/response.rs

connect_rpc/
response.rs
