Coercing HttpResponse into HttpRespnose<B>

[harr1424]

I've been scratching my head trying to figure out how to return a ServiceResponse that conforms to ServiceResponse<B>. My middleware is designed to rate limit server endpoints. The middleware will either forward the original request to the server, or response with an HTTP status code of Bad Request or Too Many Requests dependent upon some simple logic. The difficulty I am having is creating an HttpResponse<B> from an HttpResponse.

pub async fn handle_rate_limiting<S, B>(
    req: ServiceRequest,
    limiter: Arc<Mutex<Limiter>>,
    service: Arc<S>,
) -> Result<ServiceResponse<B>, Error>
where
    S: Service<ServiceRequest, Response=ServiceResponse<B>, Error=Error>,
    S::Future: 'static,
    B: 'static + MessageBody,
{
    let ip = match req.peer_addr().map(|addr| addr.ip().to_string()) {
        Some(ip) => ip,
        None => return Ok(ServiceResponse::new(req.request().clone(), HttpResponse::new(StatusCode::BAD_REQUEST))) // expected `HttpResponse<B>`, found `HttpResponse`
    };

    println!("Request from IP: {}", ip);

    let now = Utc::now();
    {
        let mut limiter = limiter.lock().unwrap();
        let (last_request_time, request_count) = limiter.ip_addresses.entry(ip.clone())
            .or_insert((now, 0));

        println!("IP: {} - Last Request Time: {}, Request Count: {}", ip, last_request_time, request_count);

        if now - *last_request_time <= Duration::seconds(20) {
            if *request_count >= 2 {
                println!("IP: {} - Too Many Requests", ip);
                return Ok(ServiceResponse::new(req.request().clone(), HttpResponse::new(StatusCode::TOO_MANY_REQUESTS))); // expected `HttpResponse<B>`, found `HttpResponse`
            } else {
                *request_count += 1;
                println!("IP: {} - Incremented Request Count: {}", ip, request_count);
            }
        } else {
            // Reset time and count after 20 seconds
            *last_request_time = now;
            *request_count = 1;
            println!("IP: {} - Reset Request Count and Time", ip);
        }
    }

    let res = service.call(req).await?;
    Ok(res)
}

I have tried using map_into_right_body() and map_into_left_body() but both of these return an EitherBody. Using into_body() appears as ig it should do just what I am trying but actual transforms the return type into a BoxBody. map_into_boxed_body does not change the type from HttpResponse at all.

Can anyone please explain what I am missing here?

Note that the answer to this similar question does not have a solution that works here. Using Response = ServiceResponse<EitherBody<B, BoxBody>> results in a similar compilation error:
expected ServiceResponse, found ServiceResponse<EitherBody>``

My full middleware definition is as follows:

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use chrono::{DateTime, Duration, Utc};
use actix_service::{Service, Transform};
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::{Error, HttpResponse};
use futures::future::{ok, Ready};
use std::task::{Context, Poll};
use actix_web::body::{BoxBody, EitherBody, MessageBody};
use actix_web::http::StatusCode;

pub struct Limiter {
    pub ip_addresses: HashMap<String, (DateTime<Utc>, usize)>,
}

pub struct RateLimiter {
    pub(crate) limiter: Arc<Mutex<Limiter>>,
}

impl RateLimiter {
    pub fn new(limiter: Arc<Mutex<Limiter>>) -> Self {
        Self { limiter }
    }
}

pub struct RateLimiterMiddleware<S> {
    pub(crate) service: Arc<S>,
    pub(crate) limiter: Arc<Mutex<Limiter>>,
}

impl<S, B> Transform<S, ServiceRequest> for RateLimiter
where
    S: Service<ServiceRequest, Response=ServiceResponse<EitherBody<B, BoxBody>>, Error=Error> + 'static,
    S::Future: 'static,
    B: 'static + MessageBody,
{
    type Response=ServiceResponse<EitherBody<B, BoxBody>>;
    type Error = Error;
    type Transform = RateLimiterMiddleware<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ok(RateLimiterMiddleware {
            service: Arc::new(service),
            limiter: self.limiter.clone(),
        })
    }
}

impl<S, B> Service<ServiceRequest> for RateLimiterMiddleware<S>
where
    S: Service<ServiceRequest, Response=ServiceResponse<EitherBody<B, BoxBody>>, Error=Error> + 'static,
    S::Future: 'static,
    B: 'static + MessageBody,
{
    type Response=ServiceResponse<EitherBody<B, BoxBody>>;
    type Error = Error;
    type Future = Pin<Box<dyn Future<Output=Result<Self::Response, Self::Error>>>>;

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

    fn call(&self, req: ServiceRequest) -> Self::Future {
        let limiter = Arc::clone(&self.limiter);
        let service = Arc::clone(&self.service);

        Box::pin(handle_rate_limiting(req, limiter, service))
    }
}

pub async fn handle_rate_limiting<S, B>(
    req: ServiceRequest,
    limiter: Arc<Mutex<Limiter>>,
    service: Arc<S>,
) -> Result<ServiceResponse<EitherBody<B, BoxBody>>, Error>
where
    S: Service<ServiceRequest, Response=ServiceResponse<EitherBody<B, BoxBody>>, Error=Error>,
    S::Future: 'static,
    B: 'static,
{
    let ip = match req.peer_addr().map(|addr| addr.ip().to_string()) {
        Some(ip) => ip,
        None => return Ok(ServiceResponse::new(req.request().clone(), HttpResponse::new(StatusCode::BAD_REQUEST).map_into_boxed_body().map_into_right_body())) //  expected `HttpResponse<EitherBody<B>>`, found `HttpResponse`
    };

    println!("Request from IP: {}", ip);

    let now = Utc::now();
    {
        let mut limiter = limiter.lock().unwrap();
        let (last_request_time, request_count) = limiter.ip_addresses.entry(ip.clone())
            .or_insert((now, 0));

        println!("IP: {} - Last Request Time: {}, Request Count: {}", ip, last_request_time, request_count);

        if now - *last_request_time <= Duration::seconds(20) {
            if *request_count >= 2 {
                println!("IP: {} - Too Many Requests", ip);
                return Ok(ServiceResponse::new(req.request().clone(), HttpResponse::new(StatusCode::TOO_MANY_REQUESTS).map_into_boxed_body().map_into_right_body()));
            } else {
                *request_count += 1;
                println!("IP: {} - Incremented Request Count: {}", ip, request_count);
            }
        } else {
            // Reset time and count after 20 seconds
            *last_request_time = now;
            *request_count = 1;
            println!("IP: {} - Reset Request Count and Time", ip);
        }
    }

    let res = service.call(req).await?;
    Ok(res)
}

I am trying to wrap my endpoints as shown below:

    let limiter = Arc::new(Mutex::new(Limiter {
        ip_addresses: HashMap::new(),
    }));

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(app_state.clone()))
            .wrap(Logger::default())
            .wrap_fn(move |req, srv| {
                let srv = Arc::new(srv);
                let limiter = Arc::clone(&limiter);
                async move {
                    handle_rate_limiting(req, limiter, srv).await
                }
            })

[robjtede]

Kinda duplicate of #3428

You can't coerce something to a B here any more than other generics in Rust.

You're on the right track using EitherBody and the helper methods. The missing piece is in the return type:

) -> Result<ServiceResponse<EitherBody<B>>, Error>

In essensce, you're wrapping a Svc(ResponseBody = B) into a Svc(ResponseBody = EitherBody<B, BoxBody>), where B is still only from the service.call(req).await result, and BoxBody is from your other responses.

[harr1424]

Thank you for the correction on the return type.

I do not understand why after making the adjustments I am still seeing a type mismatch at compile time:

pub async fn handle_rate_limiting<S, B>(
    req: ServiceRequest,
    limiter: Arc<Mutex<Limiter>>,
    service: Arc<S>,
) -> Result<ServiceResponse<EitherBody<B, BoxBody>>, Error>
where
    S: Service<ServiceRequest, Response=ServiceResponse<EitherBody<B, BoxBody>>, Error=Error>,
    S::Future: 'static,
    B: 'static,
{
    let ip = match req.peer_addr().map(|addr| addr.ip().to_string()) {
        Some(ip) => ip,
        None => return Ok(ServiceResponse::new(req.request().clone(), HttpResponse::new(StatusCode::BAD_REQUEST).map_into_boxed_body())) //  expected `HttpResponse<EitherBody<B>>`, found `HttpResponse`
    };

Why isn't the helper method leading to a return of an HttpResponse<BoxBody> ?

[robjtede]

• .map_into_boxed_body(): T -> BoxBody
• .map_into_right_body(): T -> EitherBody<_, T>

add them together:

.map_into_boxed_body().map_into_right_body() // T -> EitherBody<_, BoxBody>

[harr1424]

Thanks, for me that wasn't intuitive. Please forgive me as I am very new to Rust and trying my best here, but is the trait bound involving Response=ServiceResponse<EitherBody<B, BoxBody>> intended to be applied throughout the definition of my middleware? I've specified this trait bound for the Transform and Service implementations.

When I attempt to warp my endpoints using the middleware I see:

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(app_state.clone()))
            .wrap(Logger::default())
            .wrap_fn(move |req, srv| {
                let srv = Arc::new(srv);
                let limiter = Arc::clone(&limiter);
                async move {
                    handle_rate_limiting(req, limiter, srv).await //  expected `ServiceResponse<EitherBody<_>>`, found `ServiceResponse<StreamLog<...>>`
                }
            })

[robjtede]

Firstly, I'd recommend using middleware::from_fn() over wrap_fn; the latter will likely be removed in the next major version.

you're wrapping a Svc(ResponseBody = B)

What I meant by this is that tour bound on S should be reverted to the orginal.

-   S: Service<ServiceRequest, Response=ServiceResponse<EitherBody<B, BoxBody>>, Error=Error>,
+   S: Service<ServiceRequest, Response=ServiceResponse<B>, Error=Error>,

The intuition for middleware in general is: you know nothing about the specific body types passed to the middleware, you only know that B: MessageBody.