reqwest_connect_rpc/
client.rs

1// Copyright 2025 Anapaya Systems
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//   http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14//! Connect-RPC client library using reqwest.
15
16use std::{borrow::Cow, sync::Arc, time::Duration};
17
18use bytes::Bytes;
19use reqwest::header::{self, HeaderMap, HeaderValue};
20use thiserror::Error;
21use tracing::Instrument;
22
23use crate::{
24    error::CrpcError,
25    token_source::{TokenSource, TokenSourceError},
26};
27
28/// Connect RPC client error.
29#[derive(Debug, Error)]
30pub enum CrpcClientError {
31    /// Error that occurs when there is a connection issue.
32    #[error("connection error {context}: {source:#?}")]
33    ConnectionError {
34        /// Additional context about the connection error.
35        context: Cow<'static, str>,
36        /// The underlying source error.
37        source: Box<dyn std::error::Error + Send + Sync + 'static>,
38    },
39    /// Error returned by the server.
40    #[error("server returned an error: {0:#?}")]
41    CrpcError(CrpcError),
42    /// Error decoding the response body.
43    #[error("failed to decode response body: {context}: {source:#?}")]
44    DecodeError {
45        /// Additional context about the decoding error.
46        context: Cow<'static, str>,
47        /// The underlying source error.
48        source: Option<Box<dyn std::error::Error + Send + Sync + 'static>>,
49        /// The response body, if available.
50        body: Option<Bytes>,
51    },
52    /// Error retrieving a token from the token source.
53    #[error("failed to retrieve token: {0}")]
54    TokenSourceError(#[from] TokenSourceError),
55    /// The token cannot be sent as an `Authorization` header value.
56    #[error("failed to format token as header value: {0}")]
57    InvalidTokenHeader(#[source] header::InvalidHeaderValue),
58}
59
60impl CrpcClientError {
61    /// Returns whether the failure is transient, so that a retry may help.
62    ///
63    /// Prefer this over matching the variants: a new variant would silently fall into a caller's
64    /// wildcard arm.
65    #[must_use]
66    pub fn is_transient(&self) -> bool {
67        match self {
68            // The exchange did not complete: name resolution failed, the host has no route to
69            // the server, or the connection dropped part way through the response body.
70            Self::ConnectionError { .. } => true,
71            Self::CrpcError(error) => error.is_transient(),
72            Self::DecodeError { .. } => false,
73            Self::InvalidTokenHeader(_) => false,
74            Self::TokenSourceError(error) => error.is_transient(),
75        }
76    }
77}
78
79/// Error creating a [`CrpcClient`].
80#[derive(Debug, Error)]
81pub enum CrpcClientCreationError {
82    /// The HTTP client could not be built.
83    #[error("failed to build HTTP client")]
84    HttpClient(#[source] reqwest::Error),
85    /// The user agent cannot be sent as a `User-Agent` header value.
86    #[error("invalid user agent {user_agent:?}")]
87    InvalidUserAgent {
88        /// The rejected user agent.
89        user_agent: String,
90        /// The underlying cause.
91        #[source]
92        source: header::InvalidHeaderValue,
93    },
94}
95
96impl CrpcClientCreationError {
97    /// Returns whether the failure is transient, so that a retry may help.
98    ///
99    /// Prefer this over matching the variants: a new variant would silently fall into a caller's
100    /// wildcard arm.
101    #[must_use]
102    pub fn is_transient(&self) -> bool {
103        match self {
104            // Creating a client builds a TLS backend and formats a header value from arguments the
105            // caller supplies, so the next attempt with the same arguments fails the same way.
106            Self::HttpClient(_) | Self::InvalidUserAgent { .. } => false,
107        }
108    }
109}
110
111const APPLICATION_PROTO: &str = "application/proto";
112
113/// A Connect-RPC client.
114pub struct CrpcClient {
115    http_client: reqwest::Client,
116    base_url: url::Url,
117    token_source: Option<Arc<dyn TokenSource>>,
118    user_agent: HeaderValue,
119}
120
121impl CrpcClient {
122    /// Creates a new [`CrpcClient`] for the given base URL.
123    pub fn new(base_url: &url::Url) -> Result<Self, CrpcClientCreationError> {
124        let http_client = reqwest::ClientBuilder::new()
125            .timeout(Duration::from_secs(30))
126            .build()
127            .map_err(CrpcClientCreationError::HttpClient)?;
128
129        Self::new_with_client(base_url, http_client)
130    }
131
132    /// Creates a new [`CrpcClient`] for the given base URL and explicit [`reqwest::Client`].
133    pub fn new_with_client(
134        base_url: &url::Url,
135        http_client: reqwest::Client,
136    ) -> Result<Self, CrpcClientCreationError> {
137        let default_user_agent = format!("reqwest-crpc {}", env!("CARGO_PKG_VERSION"));
138        let user_agent = HeaderValue::from_str(&default_user_agent).map_err(|source| {
139            CrpcClientCreationError::InvalidUserAgent {
140                user_agent: default_user_agent,
141                source,
142            }
143        })?;
144
145        Ok(CrpcClient {
146            http_client,
147            base_url: base_url.clone(),
148            token_source: None,
149            user_agent,
150        })
151    }
152
153    /// Uses given token source for authentication of all following requests.
154    pub fn use_token_source(&mut self, token_source: Arc<dyn TokenSource>) -> &mut Self {
155        self.token_source = Some(token_source);
156        self
157    }
158
159    /// Sets the user agent header for all following requests.
160    pub fn use_user_agent(
161        &mut self,
162        user_agent: &str,
163    ) -> Result<&mut Self, CrpcClientCreationError> {
164        self.user_agent = HeaderValue::from_str(user_agent).map_err(|source| {
165            CrpcClientCreationError::InvalidUserAgent {
166                user_agent: user_agent.to_owned(),
167                source,
168            }
169        })?;
170        Ok(self)
171    }
172
173    /// Unary RPC request with a [`buffa`] message.
174    pub async fn unary_request<Req, Res>(
175        &self,
176        path: &str,
177        req: &Req,
178    ) -> Result<Res, CrpcClientError>
179    where
180        Req: buffa::Message,
181        Res: buffa::Message,
182    {
183        let body = self.post(path, req.encode_to_vec()).await?;
184
185        Res::decode_from_slice(&body).map_err(|e| {
186            CrpcClientError::DecodeError {
187                context: "error decoding response body".into(),
188                source: Some(e.into()),
189                body: Some(body),
190            }
191        })
192    }
193
194    /// Posts `body` to `path` and returns the response body.
195    ///
196    /// Codec agnostic: the caller encodes the request and decodes the response.
197    async fn post(&self, path: &str, body: Vec<u8>) -> Result<Bytes, CrpcClientError> {
198        self.do_post(path, body)
199            .instrument(tracing::info_span!("request", %path, id = rand::random::<u16>()))
200            .await
201    }
202
203    async fn do_post(&self, path: &str, body: Vec<u8>) -> Result<Bytes, CrpcClientError> {
204        let url = self.base_url.join(path).map_err(|e| {
205            CrpcClientError::ConnectionError {
206                context: "error joining base URL and path".into(),
207                source: e.into(),
208            }
209        })?;
210
211        let mut headers = HeaderMap::with_capacity(3);
212        headers.insert(
213            header::CONTENT_TYPE,
214            header::HeaderValue::from_static(APPLICATION_PROTO),
215        );
216        headers.insert(header::USER_AGENT, self.user_agent.clone());
217
218        tracing::trace!(?url, ?headers, "Sending crpc unary request");
219
220        if let Some(token_source) = &self.token_source {
221            let token = token_source.get_token().await?;
222            let token_header = header::HeaderValue::from_str(&token_source.format_header(token))
223                .map_err(CrpcClientError::InvalidTokenHeader)?;
224
225            headers.insert(header::AUTHORIZATION, token_header);
226        }
227
228        let response = self
229            .http_client
230            .post(url)
231            .body(reqwest::Body::from(body))
232            .headers(headers)
233            .send()
234            .await
235            .map_err(|e| {
236                CrpcClientError::ConnectionError {
237                    context: "error sending request".into(),
238                    source: e.into(),
239                }
240            })?;
241
242        tracing::trace!(status=%response.status(), body_len=%response.content_length().unwrap_or(0), "Received crpc unary response");
243
244        let status = response.status();
245        if !status.is_success() {
246            let response_raw = response
247                .text()
248                .await
249                .unwrap_or_else(|_| "<failed to read body>".to_string());
250
251            // Try to parse the body as a CrpcError, otherwise create a generic one.
252            match serde_json::from_str::<CrpcError>(&response_raw) {
253                Ok(crpc_err) => {
254                    return Err(CrpcClientError::CrpcError(crpc_err));
255                }
256                Err(_) => {
257                    return Err(CrpcClientError::CrpcError(CrpcError::new(
258                        status.into(),
259                        response_raw,
260                    )));
261                }
262            }
263        }
264
265        response.bytes().await.map_err(|e| {
266            CrpcClientError::ConnectionError {
267                context: "error reading response body".into(),
268                source: e.into(),
269            }
270        })
271    }
272}
reqwest_connect_rpc/client.rs

reqwest_connect_rpc/
client.rs
