Tower en Rust: Service, Layer y middleware componible para aplicaciones async

Tower es una biblioteca de componentes para construir aplicaciones de red robustas y componibles. Define dos traits fundamentales, Service y Layer, que son la base del sistema de middlewares en prácticamente todo el ecosistema Rust async: Axum, Tonic (gRPC), Hyper y muchos otros frameworks los usan directamente.

La idea central de Tower es que un servidor web, una llamada a base de datos o una petición HTTP son todos, conceptualmente, lo mismo: recibes una petición, haces algo, devuelves una respuesta. Eso es un Service.

El trait Service

pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
    fn call(&mut self, req: Request) -> Self::Future;
}

Un Service<Request> recibe un Request y devuelve un future que produce Result<Response, Error>. El método poll_ready permite al service indicar si está listo para aceptar más peticiones (útil para backpressure).

Un service mínimo:

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tower::Service;

struct Eco;

impl Service<String> for Eco {
    type Response = String;
    type Error = std::convert::Infallible;
    type Future = Pin<Box<dyn Future<Output = Result<String, Self::Error>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: String) -> Self::Future {
        Box::pin(async move { Ok(format!("ECO: {}", req)) })
    }
}

El trait Layer

Un Layer es una fábrica de middlewares. Envuelve un Service para añadirle comportamiento:

use tower::Layer;

pub trait Layer<S> {
    type Service;
    fn layer(&self, inner: S) -> Self::Service;
}

// Un Layer de logging que añade logging a cualquier Service
struct LogLayer {
    prefijo: &'static str,
}

struct LogService<S> {
    inner: S,
    prefijo: &'static str,
}

impl<S> Layer<S> for LogLayer {
    type Service = LogService<S>;

    fn layer(&self, inner: S) -> LogService<S> {
        LogService { inner, prefijo: self.prefijo }
    }
}

impl<S, Req> Service<Req> for LogService<S>
where
    S: Service<Req>,
    Req: std::fmt::Debug,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = S::Future;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Req) -> Self::Future {
        println!("[{}] Petición: {:?}", self.prefijo, req);
        self.inner.call(req)
    }
}

ServiceBuilder: componer capas

Tower incluye ServiceBuilder para componer múltiples capas de forma legible:

[dependencies]
tower = { version = "0.4", features = ["full"] }
tower-http = { version = "0.5", features = ["full"] }

use tower::{ServiceBuilder, ServiceExt};
use tower_http::{
    trace::TraceLayer,
    timeout::TimeoutLayer,
    compression::CompressionLayer,
    cors::CorsLayer,
};
use std::time::Duration;

let service = ServiceBuilder::new()
    .layer(TraceLayer::new_for_http())
    .layer(TimeoutLayer::new(Duration::from_secs(10)))
    .layer(CompressionLayer::new())
    .layer(CorsLayer::permissive())
    .service(mi_handler);

Las capas se aplican de afuera hacia adentro: la primera en ServiceBuilder es la más externa (la primera en ver la petición). Esto es importante para el timeout: si lo pones antes de la compresión, el timeout incluye el tiempo de comprimir la respuesta.

Middlewares de tower-http

El crate tower-http proporciona middlewares listos para usar en aplicaciones HTTP:

TraceLayer

use tower_http::trace::{TraceLayer, DefaultMakeSpan, DefaultOnResponse};
use tracing::Level;

let trace = TraceLayer::new_for_http()
    .make_span_with(DefaultMakeSpan::new().level(Level::INFO))
    .on_response(DefaultOnResponse::new().level(Level::INFO));

TimeoutLayer

use tower_http::timeout::TimeoutLayer;
use std::time::Duration;

let timeout = TimeoutLayer::new(Duration::from_secs(30));

CompressionLayer

use tower_http::compression::CompressionLayer;

// Comprime automáticamente con gzip, brotli o deflate según Accept-Encoding
let compresion = CompressionLayer::new();

ConcurrencyLimitLayer

use tower::limit::ConcurrencyLimitLayer;

// Máximo 100 peticiones simultáneas
let limite = ConcurrencyLimitLayer::new(100);

Usar Tower directamente en Axum

Axum está construido sobre Tower, así que cualquier middleware de Tower funciona en Axum con .layer():

use axum::{Router, routing::get};
use tower::ServiceBuilder;
use tower_http::trace::TraceLayer;
use tower_http::cors::CorsLayer;

let app = Router::new()
    .route("/", get(handler))
    .layer(
        ServiceBuilder::new()
            .layer(TraceLayer::new_for_http())
            .layer(CorsLayer::permissive())
    );

Crear un middleware con tower::Service manualmente

Un ejemplo de middleware que añade una cabecera personalizada a todas las respuestas:

use axum::http::{Request, Response};
use tower::{Service, Layer};
use std::{future::Future, pin::Pin, task::{Context, Poll}};

#[derive(Clone)]
struct CabeceraLayer {
    clave: &'static str,
    valor: &'static str,
}

#[derive(Clone)]
struct CabeceraService<S> {
    inner: S,
    clave: &'static str,
    valor: &'static str,
}

impl<S> Layer<S> for CabeceraLayer {
    type Service = CabeceraService<S>;

    fn layer(&self, inner: S) -> CabeceraService<S> {
        CabeceraService { inner, clave: self.clave, valor: self.valor }
    }
}

impl<S, B> Service<Request<B>> for CabeceraService<S>
where
    S: Service<Request<B>, Response = Response<axum::body::Body>> + Clone + Send + 'static,
    S::Future: Send + 'static,
    B: Send + 'static,
{
    type Response = Response<axum::body::Body>;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Request<B>) -> Self::Future {
        let clave = self.clave;
        let valor = self.valor;
        let fut = self.inner.call(req);

        Box::pin(async move {
            let mut respuesta = fut.await?;
            respuesta.headers_mut().insert(
                clave,
                valor.parse().unwrap(),
            );
            Ok(respuesta)
        })
    }
}

Tower es la base invisible de gran parte del ecosistema Rust async. Conocer sus abstracciones fundamentales permite escribir middlewares reutilizables que funcionan en Axum, en clientes HTTP, en servidores gRPC y en cualquier otro stack que use Tower.