# timvw-rust-runtime
The runtime of Rust clients generated by
[openapi-generator-timvw-rust](https://github.com/timvw/openapi-generator-timvw-rust) (`-g timvw-rust`, with the
default `httpStack=tower`). A generated client sends every request through a `Client`: a
[tower](https://docs.rs/tower) service from `http::Request<Body>` to `http::Response<Body>`. That makes everything
around sending a tower layer: retries, timeouts, logging, headers, authentication, rate limits. Use the layers this
crate ships, any layer from the tower ecosystem, or write your own.
- [Quick start](#quick-start)
- [Retries](#retries)
- [Timeouts](#timeouts)
- [Logging](#logging)
- [Middleware: headers, authentication, layers of your own](#middleware)
- [Concurrency and rate limits](#concurrency-and-rate-limits)
- [Choosing the HTTP client, and tests without a network](#choosing-the-http-client)
- [Your own types, and streams of server-sent events](#your-own-types-and-server-sent-events)
- [Errors](#errors)
- [Features](#features)
## Quick start
A generated client keeps its `Client` in `Configuration.client`, which starts as `Client::default()`: reqwest, with
[retries](#retries) of `429`, `408`, `5xx` and failed connections. `layer` wraps it in more:
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::Client;
use tower::timeout::TimeoutLayer;
let client = Client::default() // reqwest, with retries
.layer(TraceLayer::new()) // logs each call, and each retry
.layer(TimeoutLayer::new(Duration::from_secs(120))); // bounds the whole call, retries included
// In a generated client:
// let mut configuration = my_client::apis::configuration::Configuration::new();
// configuration.client = client;
```
**Layers apply from the inside out: the last one added sees a request first.** Layers added to `Client::default()`
wrap its retries. To put a layer inside the retries (a timeout or a log line per attempt), or to retry differently,
build the stack from the bare service, `Client::from_service(ReqwestService::default())`:
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::retry::RetryPolicy;
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::{Client, ReqwestService};
use tower::retry::RetryLayer;
use tower::timeout::TimeoutLayer;
let client = Client::from_service(ReqwestService::default()) // reqwest, no retries
.layer(TimeoutLayer::new(Duration::from_secs(60))) // each attempt
.layer(TraceLayer::new()) // logs each attempt
.layer(RetryLayer::new(RetryPolicy::default())); // retries
```
A `Client` is cheap to clone, and clones share their layers' state (a rate limit, a token cache).
## Retries
`Client::default()`, `Client::new()` and `Client::from_reqwest(..)` retry with `RetryPolicy::default()` (feature
`retry`, on by default). `retry::RetryPolicy` is a `tower::retry::Policy` for HTTP APIs:
- It retries `429 Too Many Requests`, `408 Request Timeout`, every `5xx` and failures to connect. Any other response
or error is returned at once.
- It waits as long as `Retry-After` asks (in seconds), or else backs off exponentially: 200 ms, 400 ms, 800 ms, ...,
capped at 8 s. The cap applies to `Retry-After` too.
- It retries at most 3 times by default.
- It does not retry a request whose body is a stream (a streamed file upload), since that cannot be sent twice. JSON,
form, text and in-memory multipart bodies are sent again.
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::retry::RetryPolicy;
use timvw_rust_runtime::{Client, ReqwestService};
use tower::retry::RetryLayer;
let policy = RetryPolicy::new(5)
.base_delay(Duration::from_millis(500))
.max_delay(Duration::from_secs(30));
let client = Client::from_service(ReqwestService::default()).layer(RetryLayer::new(policy));
```
Without retries: `Client::from_service(ReqwestService::default())`, or build without the `retry` feature.
The policy retries every method, `POST` included: for most APIs a `429` or `503` means the request was not processed.
If a retried `POST` could act twice on your API, send an idempotency key the API understands, or write a
`tower::retry::Policy` of your own: the trait has two methods, `retry` (whether to try again and after which delay)
and `clone_request`. `Body::try_clone` clones a body held in memory.
## Timeouts
`tower::timeout::TimeoutLayer` fails a request that takes too long. Where it sits decides what it bounds (see
[Quick start](#quick-start)). A timeout is an `Error` whose `is_timeout()` is true:
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::{Client, Method};
async fn list_items(client: &Client) -> Result<(), timvw_rust_runtime::Error> {
match client.request(Method::GET, "https://api.example.com/items").send().await {
Err(error) if error.is_timeout() => eprintln!("too slow, try again later"),
response => println!("{}", response?.status()),
}
Ok(())
}
let client = Client::default().layer(tower::timeout::TimeoutLayer::new(Duration::from_secs(30)));
```
reqwest's own timeouts (connect timeout, read timeout) are set on a `reqwest::Client`, see
[Choosing the HTTP client](#choosing-the-http-client). For a response streamed for minutes (server-sent events), a
total timeout ends the stream; use a connect timeout instead, or none.
## Logging
`trace::TraceLayer` (feature `tracing`) logs every request with [`tracing`](https://docs.rs/tracing):
- a `http.request` span with `http.request.method`, `url.full` (without the query, which can hold an API key) and
`http.response.status_code`;
- an event when the response arrives, with `status` and `elapsed_ms`: `DEBUG` below `500`, `WARN` for a `5xx` or a
failure to send;
- with `RetryPolicy`, a `DEBUG` event `retrying` with `attempt`, `max_retries`, `delay_ms` and `status`.
Headers and bodies are never logged. The target is `timvw_rust_runtime::trace`:
```rust,no_run
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::Client;
// RUST_LOG=timvw_rust_runtime=debug shows every request, RUST_LOG=timvw_rust_runtime=warn only failures.
tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
.init();
let client = Client::default().layer(TraceLayer::new());
```
```text
WARN http.request{http.request.method=POST url.full=https://api.example.com/v1/items http.response.status_code=503}: timvw_rust_runtime::trace: response status=503 elapsed_ms=41
DEBUG timvw_rust_runtime::trace: retrying attempt=1 max_retries=3 delay_ms=200 status=503
DEBUG http.request{http.request.method=POST url.full=https://api.example.com/v1/items http.response.status_code=200}: timvw_rust_runtime::trace: response status=200 elapsed_ms=380
```
reqwest's and hyper's own `DEBUG` events (connecting, pooling) are logged inside the `http.request` span too.
For more (headers, custom spans and fields, OpenTelemetry), use [`tower-http`](https://docs.rs/tower-http)'s
`TraceLayer`; it fits the same way:
```rust,no_run
let client = timvw_rust_runtime::Client::default().layer(tower_http::trace::TraceLayer::new_for_http());
```
## Middleware
Anything that changes a request or a response is a layer. Three ways to write one, from the shortest:
**A header on every request**: `tower::util::MapRequestLayer` with a function, or tower-http's
`SetRequestHeaderLayer`:
```rust,no_run
use timvw_rust_runtime::http::{self, HeaderName, HeaderValue};
use timvw_rust_runtime::{Body, Client};
let client = Client::default()
.layer(tower::util::MapRequestLayer::new(|mut request: http::Request<Body>| {
request.headers_mut().insert("x-team", HeaderValue::from_static("search"));
request
}))
.layer(tower_http::set_header::SetRequestHeaderLayer::if_not_present(
HeaderName::from_static("x-request-source"),
HeaderValue::from_static("batch"),
));
```
**Several layers at once**: `tower::ServiceBuilder` composes layers top to bottom (the first one sees a request
first), the reverse of chaining `Client::layer`:
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::retry::RetryPolicy;
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::{Client, ReqwestService};
let client = Client::from_service(ReqwestService::default()).layer(
tower::ServiceBuilder::new()
.retry(RetryPolicy::default())
.layer(TraceLayer::new())
.timeout(Duration::from_secs(60)),
);
```
**A layer of your own**, for work that needs state or `await`, such as fetching and caching an access token. A layer
makes the service; the service changes the request and calls the service inside it:
```rust,no_run
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use timvw_rust_runtime::http::{self, header::AUTHORIZATION, HeaderValue};
use timvw_rust_runtime::{Body, Client};
use tower::{Layer, Service};
#[derive(Clone)]
struct AccessToken {
cached: Arc<tokio::sync::Mutex<Option<String>>>,
}
impl<S> Layer<S> for AccessToken {
type Service = WithAccessToken<S>;
fn layer(&self, inner: S) -> Self::Service {
WithAccessToken { inner, cached: self.cached.clone() }
}
}
#[derive(Clone)]
struct WithAccessToken<S> {
inner: S,
cached: Arc<tokio::sync::Mutex<Option<String>>>,
}
impl<S> Service<http::Request<Body>> for WithAccessToken<S>
where
S: Service<http::Request<Body>> + Clone + Send + 'static,
S::Future: Send,
{
type Response = S::Response;
type Error = S::Error;
type Future = Pin<Box<dyn Future<Output = Result<S::Response, S::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, mut request: http::Request<Body>) -> Self::Future {
// Call the service that was polled ready, and keep a fresh clone for the next request.
let fresh = self.inner.clone();
let mut inner = std::mem::replace(&mut self.inner, fresh);
let cached = self.cached.clone();
Box::pin(async move {
let mut cached = cached.lock().await;
if cached.is_none() {
*cached = Some(fetch_token().await);
}
let value = HeaderValue::from_str(&format!("Bearer {}", cached.as_deref().unwrap())).unwrap();
drop(cached);
request.headers_mut().insert(AUTHORIZATION, value);
inner.call(request).await
})
}
}
async fn fetch_token() -> String {
"token".to_owned() // ask your identity provider
}
let client = Client::default().layer(AccessToken { cached: Arc::default() });
```
A layer's service must be `Clone + Send`; it need not be `Sync`. Its response body can be any `http_body::Body`.
## Concurrency and rate limits
`tower::limit::ConcurrencyLimitLayer` caps requests in flight; the clones of a `Client` share the cap.
`RateLimit` cannot be cloned, so put a `Buffer` in front of it: the buffer shares one limiter between all clones. A
buffer starts a tokio task, so build that client inside a tokio runtime.
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::Client;
#[tokio::main]
async fn main() {
let client = Client::default()
.layer(tower::limit::ConcurrencyLimitLayer::new(8))
.layer(tower::ServiceBuilder::new().buffer(256).rate_limit(50, Duration::from_secs(1)));
}
```
These layers need tower's `limit` and `buffer` features in your `Cargo.toml`, like `timeout` for `TimeoutLayer`:
`tower = { version = "0.5", features = ["limit", "buffer", "timeout"] }`.
## Choosing the HTTP client
`Client::from_reqwest` sends with a `reqwest::Client` you built (a proxy, a custom TLS setup, connect and read timeouts,
pool settings), with the default retries; `Client::from_service(ReqwestService::new(reqwest_client))` without them:
```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::Client;
let reqwest_client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(5))
.proxy(reqwest::Proxy::all("http://proxy.internal:3128").unwrap())
.build()
.unwrap();
let client = Client::from_reqwest(reqwest_client);
```
`Client::from_service` sends with any tower service from `http::Request<Body>` to `http::Response<B>`, as is, with no
retries added: a hyper client, a client for another runtime, or a function that answers without a network, for tests:
```rust,no_run
use timvw_rust_runtime::{http, Body, BoxError, Client};
let page = r#"{"data":[{"id":"1","name":"Lamp"}],"has_more":false}"#;
Ok::<_, BoxError>(
http::Response::builder()
.header("content-type", "application/json")
.body(Body::from(page))?,
)
}));
```
Without the `reqwest` feature (and `rustls` or `native-tls`), the crate does not depend on reqwest, and
`Client::default()` fails every request: set a service with `Client::from_service`.
## Your own types and server-sent events
A generated function decodes to the spec's models. To send and decode types of your own (fewer fields, a provider
whose replies differ, `serde_json::Value`), call the client directly: the same layers apply (retries, logging).
`Response::events()` reads a `text/event-stream` body, and `.json::<T>()` decodes each event's `data:`:
```rust,no_run
use futures_util::StreamExt;
use serde::Deserialize;
use timvw_rust_runtime::{Client, Method};
#[derive(Deserialize)]
struct Delta {
id: String,
}
async fn run(client: &Client, api_key: &str) -> Result<(), timvw_rust_runtime::Error> {
let request = serde_json::json!({"model": "some-model", "stream": true});
let response = client
.request(Method::POST, "https://api.example.com/v1/chat/completions")
.bearer_auth(api_key)
.json(&request)
.send()
.await?;
let mut deltas = response.events().json::<Delta>();
while let Some(delta) = deltas.next().await {
println!("{}", delta?.id);
}
Ok(())
}
```
`data: [DONE]` or the end of the body ends the stream. A payload that does not decode yields an error whose
`is_decode()` is true, and the stream goes on. `events()` alone gives each `sse::Event` undecoded (`event:`, `data:`,
`id:`), and `sse::Decoder` is the parser, for a body read some other way. A generated client's typed event streams use
the same parser.
## Errors
A request that fails to send is an `Error`:
| `is_builder()` | the request could not be built (an invalid URL, header or body) |
| `is_request()` | the service failed; `is_timeout()` and `is_connect()` narrow it down |
| `is_body()` | reading the response body failed |
| `is_decode()` | the response could not be decoded (JSON) |
`source()` is the error of the service or layer that failed: a `reqwest::Error`, tower's `Elapsed`, or your own. A
generated client wraps it as `apis::Error::Http(error)`. A response with an error status is not an `Error` here; the
generated client turns it into `apis::Error::ResponseError`.
## Features
| `rustls` | yes | reqwest with rustls, as `Client::default()` |
| `native-tls` | no | reqwest with the platform's TLS |
| `reqwest` | via `rustls` or `native-tls` | `ReqwestService`, `Client::from_reqwest` |
| `retry` | yes | `retry::RetryPolicy` (sleeps with tokio), and retries in `Client::default()` |
| `tracing` | no | `trace::TraceLayer`, and a log event per retry |
A generated crate forwards `native-tls`, `rustls`, `retry` and `tracing` under the same names.
## What the generated code uses
The generated functions build requests with `RequestBuilder` (`query`, `header`, `bearer_auth`, `basic_auth`, `json`,
`form`, `body`, `multipart`, `build`) and read responses with `Response` (`status`, `headers`, `text`, `bytes`,
`bytes_stream`): the methods of reqwest's types of the same names. `multipart::Form` encodes multipart bodies.
## License
MIT OR Apache-2.0.