Futures error with method named `then` in a reverse proxy

help
1

I am developing a reverse proxy named palantir. It works fine and outperforms nginx reverse proxy in benchmarks. I am actually trying to migrate from actix-web v0.7.18 to built-in rust async. I have some problems which I think are because of futures-preview = { version = "=0.3.0-alpha.17", features = ["compat"] }. I am quite new to rust and the recent versions of hyper and future. I will be grateful if you could help me.
It is the the revised version of my code:

in Cargo.toml:
futures-preview = { version = "=0.3.0-alpha.17", features = ["compat"] }
hyper = "0.12.9"

proxy.rs:

    use futures::future::{self, Future};
    use {hyper::{Body, Request, Response, Client, Uri, StatusCode}};
    use std::str::FromStr;

    type BoxFut = Box<dyn Future<Output=Response<Body>> + Send>;

    fn forward_uri<B>(forward_url: &str, req: &Request<B>) -> Uri {
        let forward_uri = match req.uri().query() {
            Some(query) => format!("{}{}?{}", forward_url, req.uri().path(), query),
            None => format!("{}{}", forward_url, req.uri().path()),
        };

        Uri::from_str(forward_uri.as_str()).unwrap()
    }

    pub fn call(forward_url: &str, mut request: Request<Body>) -> BoxFut {
        *request.uri_mut() = forward_uri(forward_url, &request);
    	let proxied_request = request;
    	let client = Client::new();
        use futures::future::Future;
    	let response = client.request(proxied_request).then(|response| {

    		let proxied_response = match response {
                Ok(response) => response,
                Err(_) => {
                    // println!("Error: {}", error); // TODO: Configurable logging
                    Response::builder()
                        .status(StatusCode::INTERNAL_SERVER_ERROR)
                        .body(Body::empty())
                        .unwrap()
                },
            };


            future::ok(proxied_response)
    	});

    	Box::new(response)
    }

First, I receive this error for then in the call function:
no method named then found for type hyper::client::ResponseFuture in the current scope

Second, I cannot use Error=hyper::Error in BoxFut type, because of this error:
associated type Error not found for futures::Future

This code works fine in the crate hyper-reverse-proxy which uses hyper = "0.12" and futures = "0.1".

2
  • then is a method of futures::FutureExt, not Future, so you have to import that trait to use the method
  • You should use Future<Output = Result<T, E>> when you want to express a fallible future
    • In that case, you might want to use map_ok and map_err methods of futures::TryFutureExt trait instead of then
3

Thank you dear @tesaguri for your help. I tried to correct it by:

    use {
        hyper::{
            Body, Client, Request, Response, Server, Uri,
            service::service_fn,
            rt::run
        },
        futures::{
            compat::Future01CompatExt,
            future::{FutureExt, TryFutureExt},
        },
        std::net::SocketAddr,
        std::str::FromStr
    };

    fn forward_uri<B>(forward_url: &str, req: &Request<B>) -> Uri {
        
        let forward_uri = match req.uri().query() {
            Some(query) => format!("{}{}?{}", forward_url, req.uri().path(), query),
            None => format!("{}{}", forward_url, req.uri().path()),
        };

        Uri::from_str(forward_uri.as_str()).unwrap()
    }

    async fn call(forward_url: &str, mut _req: Request<Body>) -> Result<Response<Body>, hyper::Error> {

        let url_str = forward_uri(forward_url, &_req);
        let res = Client::new().get(url_str).compat().await;
        res
    }

    async fn run_server(forward_url: &'static str, addr: SocketAddr) {

        let serve_future = Server::bind(&addr)

            .serve(|| service_fn(|req| call(forward_url: &'static str, req).boxed().compat()));

        if let Err(e) = serve_future.compat().await {
            eprintln!("server error: {}", e);
        }
    }

    fn main() {

        let addr = SocketAddr::from(([127, 0, 0, 1], 3000));

        let futures_03_future = run_server("https://127.0.0.1:9061", addr);
        let futures_01_future = futures_03_future.unit_error().boxed().compat();

        run(futures_01_future);
    }

However, I have problems with forward_url: &'static str in the run_server function:

type ascription is experimental
  --> src/main.rs:37:41
   |
37 |         .serve(|| service_fn(|req| call(forward_url: &'static str, req).boxed().compat()));
   |                                         ^^^^^^^^^^^^^^^^^^^^^^^^^
   |
   = note: for more information, see https://github.com/rust-lang/rust/issues/23416
   = help: add `#![feature(type_ascription)]` to the crate attributes to enable

When I add #![feature(type_ascription)] to the crate attributes, I receive:

`forward_url` does not live long enough
  --> src/main.rs:37:41
   |
37 |         .serve(|| service_fn(|req| call(forward_url: &'static str, req).boxed().compat()));
   |                -- ----------------------^^^^^^^^^^^--------------------------------------
   |                |  |                     |
   |                |  |                     borrowed value does not live long enough
   |                |  returning this value requires that `forward_url` is borrowed for `'static`
   |                value captured here
...
42 | }
   | - `forward_url` dropped here while still borrowed

I will be grateful if you tell me your ideas on how to fix it.

4

You don't need to (and indeed, can't) annotate types when calling a method - change this:

.serve(|| service_fn(|req| call(forward_url: &'static str, req).boxed().compat()));

To this:

.serve(|| service_fn(|req| call(forward_url, req).boxed().compat()));

Perhaps what you meant to do was change the signature of call from async fn call(forward_url: &str, ...) to async fn call(forward_url: &'static str, ...), to match your run_server function?

5

Thank you @17cupsofcoffee for pointing this out. Actually, I had used your suggestion before, but I received forward_url does not live long enough:

`forward_url` does not live long enough
  --> src/main.rs:37:41
   |
37 |         .serve(|| service_fn(|req| call(forward_url, req).boxed().compat()));
   |                -- ----------------------^^^^^^^^^^^------------------------
   |                |  |                     |
   |                |  |                     borrowed value does not live long enough
   |                |  returning this value requires that `forward_url` is borrowed for `'static`
   |                value captured here
...
42 | }
   | - `forward_url` dropped here while still borrowed

Also, I changed the types as:
fn forward_uri<B>(forward_url: &'static str, req: &Request<B>) -> Uri {
and
async fn call(forward_url: &'static str, mut _req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
and
async fn run_server(forward_url: &'static str, addr: SocketAddr) {

6

Firstly, I corrected it by hard-coding the forward_url which seemed not to be a big deal because it will be pasted later using clap from a config.toml. But, by using move keyword one can force the closure to take the ownership of the forward_url:


use {
    hyper::{
        Body, Client, Request, Response, Server, Uri,`
        service::service_fn,
        rt::run
    },
    futures::{
        compat::Future01CompatExt,
        future::{FutureExt, TryFutureExt},
    },
    std::net::SocketAddr,
    std::str::FromStr
};

fn forward_uri<B>(forward_url: &'static str, req: &Request<B>) -> Uri {
    
    let forward_uri = match req.uri().query() {
        Some(query) => format!("{}{}?{}", forward_url, req.uri().path(), query),
        None => format!("{}{}", forward_url, req.uri().path()),
    };

    Uri::from_str(forward_uri.as_str()).unwrap()
}

async fn call(forward_url: &'static str, mut _req: Request<Body>) -> 
    Result<Response<Body>, hyper::Error> {
        *_req.uri_mut() = forward_uri(forward_url, &_req);
        let url_str = forward_uri(forward_url, &_req);
        let res = Client::new().get(url_str).compat().await;
        res
}

async fn run_server(forward_url: &'static str, addr: SocketAddr) {
    let forwarded_url = forward_url;
    let serve_future = Server::bind(&addr)
        .serve(|| service_fn(move |req| call(forwarded_url, req).boxed().compat()));
    if let Err(e) = serve_future.compat().await {
        eprintln!("server error: {}", e);
    }
}

fn main() {

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    let url = "http://127.0.0.1:9061";
    let futures_03_future = run_server(url, addr);
    let futures_01_future = futures_03_future.unit_error().boxed().compat();
    run(futures_01_future);
}

I look forward to having hyper becoming compatible with futures 0.3 to improve the performance of the proxy. I will continue it as a branch on my development fork of palantir.

Thank you @tesaguri and @17cupsofcoffee for your helpful comments :pray: :sunflower:.

7

Hyper is in fact already compatible with futures 0.3: Just use the 0.13.0-alpha.1 version.

8

@alice Thank you :pray: :cherry_blossom:, but hyper = "0.13.0-alpha.1" has a different interface for some structs. For instance, hyper::rt::run is not the same, also I face lots of ... trait is not satisfied/implemented ... errors. However, I am quite sure by immersing myself in its documentation, I am going to be able to migrate to hyper = "0.13.0-alpha.1"

9

It could have been solved by move keyword to force the closure to take ownership of forward_url. It's textbook :stuck_out_tongue_winking_eye: and am sorry for it.
.serve(|| service_fn(move |req| call(forwarded_url, req).boxed().compat()))