use std::time::Duration;
use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue, USER_AGENT};
use reqwest::{Method, RequestBuilder};
use serde::Serialize;
use serde::de::DeserializeOwned;
use crate::auth::Credentials;
use crate::config::{BaseUrl, RetryConfig, user_agent};
use crate::error::{ApiError, Error, Result};
const MAX_ERROR_BODY: usize = 2048;
pub(crate) fn retry_after(headers: &HeaderMap) -> Option<Duration> {
headers
.get(reqwest::header::RETRY_AFTER)?
.to_str()
.ok()?
.trim()
.parse::<u64>()
.ok()
.map(Duration::from_secs)
}
fn replay_is_safe(method: &Method, replay: Replay, status: u16) -> bool {
if status == 429 {
return true;
}
replay == Replay::ByMethod && method.is_idempotent()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Replay {
ByMethod,
Never,
}
#[derive(Debug, Clone, Copy, Serialize)]
pub struct Empty;
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct RestConfig {
pub base_url: String,
pub api_version: String,
pub retry: RetryConfig,
pub timeout: Option<Duration>,
}
impl RestConfig {
pub fn new(base_url: impl Into<String>) -> Self {
Self {
base_url: base_url.into(),
api_version: "v2".to_owned(),
retry: RetryConfig::default(),
timeout: None,
}
}
#[must_use]
pub fn api_version(mut self, version: impl Into<String>) -> Self {
self.api_version = version.into();
self
}
#[must_use]
pub fn retry(mut self, retry: RetryConfig) -> Self {
self.retry = retry;
self
}
#[must_use]
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = Some(timeout);
self
}
}
impl From<BaseUrl> for RestConfig {
fn from(url: BaseUrl) -> Self {
Self::new(url.as_str())
}
}
#[derive(Debug, Clone)]
pub struct RestClient {
http: reqwest::Client,
config: RestConfig,
}
impl RestClient {
pub fn new(credentials: &Credentials, config: RestConfig) -> Result<Self> {
Self::build(Some(credentials), config)
}
pub fn unauthenticated(config: RestConfig) -> Result<Self> {
Self::build(None, config)
}
fn build(credentials: Option<&Credentials>, config: RestConfig) -> Result<Self> {
let mut headers = HeaderMap::new();
if let Some(credentials) = credentials {
credentials.apply(&mut headers)?;
}
headers.insert(USER_AGENT, HeaderValue::from_static(user_agent()));
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
let mut builder = reqwest::Client::builder()
.default_headers(headers)
.redirect(reqwest::redirect::Policy::none());
if let Some(timeout) = config.timeout {
builder = builder.timeout(timeout);
}
Ok(Self {
http: builder.build().map_err(Error::transport)?,
config,
})
}
#[must_use]
pub fn config(&self) -> &RestConfig {
&self.config
}
#[must_use]
pub fn at_version(&self, api_version: &str) -> Self {
Self {
http: self.http.clone(),
config: RestConfig {
api_version: api_version.to_owned(),
..self.config.clone()
},
}
}
pub async fn get<T, Q>(&self, path: &str, query: &Q) -> Result<T>
where
T: DeserializeOwned,
Q: Serialize + ?Sized,
{
self.decode(
path,
self.send(
Method::GET,
Replay::ByMethod,
path,
Some(query),
None::<&Empty>,
)
.await?,
)
}
pub async fn delete<T, Q>(&self, path: &str, query: &Q) -> Result<T>
where
T: DeserializeOwned,
Q: Serialize + ?Sized,
{
self.decode(
path,
self.send(
Method::DELETE,
Replay::ByMethod,
path,
Some(query),
None::<&Empty>,
)
.await?,
)
}
pub async fn delete_effectful<T, Q>(&self, path: &str, query: &Q) -> Result<T>
where
T: DeserializeOwned,
Q: Serialize + ?Sized,
{
self.decode(
path,
self.send(
Method::DELETE,
Replay::Never,
path,
Some(query),
None::<&Empty>,
)
.await?,
)
}
pub async fn post<T, B>(&self, path: &str, body: &B) -> Result<T>
where
T: DeserializeOwned,
B: Serialize + ?Sized,
{
self.decode(
path,
self.send(
Method::POST,
Replay::ByMethod,
path,
None::<&Empty>,
Some(body),
)
.await?,
)
}
pub async fn put<T, B>(&self, path: &str, body: &B) -> Result<T>
where
T: DeserializeOwned,
B: Serialize + ?Sized,
{
self.decode(
path,
self.send(
Method::PUT,
Replay::ByMethod,
path,
None::<&Empty>,
Some(body),
)
.await?,
)
}
pub async fn patch<T, B>(&self, path: &str, body: &B) -> Result<T>
where
T: DeserializeOwned,
B: Serialize + ?Sized,
{
self.decode(
path,
self.send(
Method::PATCH,
Replay::ByMethod,
path,
None::<&Empty>,
Some(body),
)
.await?,
)
}
pub async fn request<T, Q, B>(
&self,
method: Method,
replay: Replay,
path: &str,
query: Option<&Q>,
body: Option<&B>,
) -> Result<T>
where
T: DeserializeOwned,
Q: Serialize + ?Sized,
B: Serialize + ?Sized,
{
self.decode(path, self.send(method, replay, path, query, body).await?)
}
pub async fn get_bytes<Q>(&self, path: &str, query: &Q) -> Result<Vec<u8>>
where
Q: Serialize + ?Sized,
{
let request = self.http.get(self.url(path)).query(query);
self.execute_bytes(&Method::GET, Replay::ByMethod, request, path)
.await
}
pub async fn request_raw<Q, B>(
&self,
method: Method,
replay: Replay,
path: &str,
query: Option<&Q>,
body: Option<&B>,
) -> Result<String>
where
Q: Serialize + ?Sized,
B: Serialize + ?Sized,
{
self.send(method, replay, path, query, body).await
}
fn url(&self, path: &str) -> String {
format!(
"{}/{}{path}",
self.config.base_url.trim_end_matches('/'),
self.config.api_version
)
}
async fn send<Q, B>(
&self,
method: Method,
replay: Replay,
path: &str,
query: Option<&Q>,
body: Option<&B>,
) -> Result<String>
where
Q: Serialize + ?Sized,
B: Serialize + ?Sized,
{
let mut request = self.http.request(method.clone(), self.url(path));
if let Some(query) = query {
request = request.query(query);
}
if let Some(body) = body {
request = request.json(body);
}
self.execute(&method, replay, request, path).await
}
async fn execute(
&self,
method: &Method,
replay: Replay,
request: RequestBuilder,
path: &str,
) -> Result<String> {
self.execute_response(method, replay, request, path)
.await?
.text()
.await
.map_err(Error::transport)
}
async fn execute_bytes(
&self,
method: &Method,
replay: Replay,
request: RequestBuilder,
path: &str,
) -> Result<Vec<u8>> {
Ok(self
.execute_response(method, replay, request, path)
.await?
.bytes()
.await
.map_err(Error::transport)?
.to_vec())
}
async fn execute_response(
&self,
method: &Method,
replay: Replay,
request: RequestBuilder,
path: &str,
) -> Result<reqwest::Response> {
let retry = &self.config.retry;
let total_attempts = retry.attempts.saturating_add(1);
let mut current = request;
for attempt in 1..=total_attempts {
let next = if attempt < total_attempts {
current.try_clone()
} else {
None
};
let response = current.send().await.map_err(Error::transport)?;
let status = response.status().as_u16();
if response.status().is_success() {
return Ok(response);
}
let retry_after = retry_after(response.headers());
let body = response.text().await.unwrap_or_default();
let api_error = ApiError::from_body(status, path, body);
let last_attempt = attempt == total_attempts;
if !retry.should_retry(status) || !replay_is_safe(method, replay, status) {
return Err(Error::Api(api_error));
}
if last_attempt {
return Err(Error::RetriesExhausted {
attempts: total_attempts,
last: api_error,
});
}
let Some(next) = next else {
return Err(Error::InvalidRequest(
"request cannot be retried because its body is a stream".to_owned(),
));
};
current = next;
let delay = retry_after.map_or_else(
|| retry.delay(attempt),
|after| after.min(retry.retry_after_cap()),
);
tracing::debug!(
path,
status,
attempt,
total_attempts,
delay_ms = delay.as_millis(),
honored_retry_after = retry_after.is_some(),
"retryable response, backing off"
);
tokio::time::sleep(delay).await;
}
unreachable!("retry loop exited without returning")
}
fn decode<T: DeserializeOwned>(&self, path: &str, body: String) -> Result<T> {
let source = if body.trim().is_empty() {
"null"
} else {
&body
};
serde_json::from_str(source).map_err(|source| Error::Decode {
path: path.to_owned(),
body: truncate(&body),
source,
})
}
}
pub(crate) fn truncate(body: &str) -> String {
if body.len() <= MAX_ERROR_BODY {
return body.to_owned();
}
let mut end = MAX_ERROR_BODY;
while !body.is_char_boundary(end) {
end -= 1;
}
format!("{}… ({} bytes total)", &body[..end], body.len())
}
#[cfg(test)]
mod tests {
use super::*;
fn client(base_url: &str) -> RestClient {
let creds = Credentials::new("key", "secret").unwrap();
RestClient::new(&creds, RestConfig::new(base_url)).unwrap()
}
fn with_retry_after(value: &str) -> HeaderMap {
let mut headers = HeaderMap::new();
headers.insert(
reqwest::header::RETRY_AFTER,
HeaderValue::from_str(value).unwrap(),
);
headers
}
#[test]
fn retry_after_reads_delta_seconds() {
assert_eq!(
retry_after(&with_retry_after("120")),
Some(Duration::from_secs(120))
);
assert_eq!(
retry_after(&with_retry_after(" 3 ")),
Some(Duration::from_secs(3))
);
}
#[test]
fn no_retry_after_header_means_the_curve_applies() {
assert_eq!(retry_after(&HeaderMap::new()), None);
}
#[test]
fn the_http_date_form_reads_as_absent() {
assert_eq!(
retry_after(&with_retry_after("Fri, 31 Dec 1999 23:59:59 GMT")),
None
);
}
#[test]
fn a_value_that_is_not_a_count_of_seconds_reads_as_absent() {
assert_eq!(retry_after(&with_retry_after("soon")), None);
assert_eq!(retry_after(&with_retry_after("-5")), None);
assert_eq!(retry_after(&with_retry_after("1.5")), None);
}
#[test]
fn url_joins_base_version_and_path() {
let client = client("https://paper-api.alpaca.markets");
assert_eq!(
client.url("/orders"),
"https://paper-api.alpaca.markets/v2/orders"
);
}
#[test]
fn trailing_slash_on_base_url_does_not_double_up() {
let client = client("https://paper-api.alpaca.markets/");
assert_eq!(
client.url("/orders"),
"https://paper-api.alpaca.markets/v2/orders"
);
}
#[test]
fn api_version_is_overridable() {
let creds = Credentials::new("key", "secret").unwrap();
let client = RestClient::new(
&creds,
RestConfig::new("https://broker-api.sandbox.alpaca.markets").api_version("v1"),
)
.unwrap();
assert_eq!(
client.url("/accounts"),
"https://broker-api.sandbox.alpaca.markets/v1/accounts"
);
}
#[test]
fn empty_body_decodes_as_unit() {
let client = client("https://paper-api.alpaca.markets");
client.decode::<()>("/orders/1", String::new()).unwrap();
client.decode::<()>("/orders/1", " ".to_owned()).unwrap();
}
#[test]
fn empty_body_decodes_as_none() {
let client = client("https://paper-api.alpaca.markets");
let decoded: Option<u8> = client.decode("/orders/1", String::new()).unwrap();
assert_eq!(decoded, None);
}
#[test]
fn decode_failure_carries_the_body() {
let client = client("https://paper-api.alpaca.markets");
let err = client
.decode::<u8>("/account", "not json".to_owned())
.unwrap_err();
match err {
Error::Decode { path, body, .. } => {
assert_eq!(path, "/account");
assert_eq!(body, "not json");
}
other => panic!("expected Decode, got {other:?}"),
}
}
#[test]
fn long_bodies_are_truncated_on_a_char_boundary() {
let body = "€".repeat(1000);
assert!(
!body.is_char_boundary(MAX_ERROR_BODY),
"the test needs a cutoff that is NOT already on a boundary"
);
let truncated = truncate(&body);
assert!(truncated.ends_with("bytes total)"));
assert!(truncated.len() < body.len());
}
#[tokio::test]
async fn a_saturating_attempt_count_still_sends_one_request() {
use wiremock::matchers::{method as m, path as p};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(m("GET"))
.and(p("/v2/account"))
.respond_with(ResponseTemplate::new(200).set_body_string("7"))
.expect(1)
.mount(&server)
.await;
let creds = Credentials::new("key", "secret").unwrap();
let client = RestClient::new(
&creds,
RestConfig::new(server.uri()).retry(RetryConfig::default().attempts(u32::MAX)),
)
.unwrap();
let got: u8 = client.get("/account", &Empty).await.unwrap();
assert_eq!(got, 7);
}
}