timvw-rust-runtime

The runtime of Rust clients generated by openapi-generator-timvw-rust (-g timvw-rust, with the default httpStack=tower). A generated client sends every request through a Client: a 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
• Retries
• Timeouts
• Logging
• Middleware: headers, authentication, layers of your own
• Concurrency and rate limits
• Choosing the HTTP client, and tests without a network
• Your own types, and streams of server-sent events
• Errors
• Features

Quick start

A generated client keeps its Client in Configuration.client, which starts as Client::default(): reqwest, with retries of 429, 408, 5xx and failed connections. layer wraps it in more:

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()):

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.

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). A timeout is an Error whose is_timeout() is true:

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. 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:

• 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:

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());

 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's TraceLayer; it fits the same way:

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:

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:

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:

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.

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:

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:

use timvw_rust_runtime::{http, Body, BoxError, Client};

let client = Client::from_service(tower::service_fn(|request: http::Request<Body>| async move {
    assert_eq!(request.uri().path(), "/v1/items");
    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::

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:

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

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.