Connecting different async code with tokio-tungstenite

[Relrin]

Hello!

I'm trying to implement checks for a token, that was specified in headers, but before doing an actual work I'd like to send a request to an existing Redis instance via an additional future (with redis-async crate at this case). The prepared future is invoking inside of the auth_middleware_callback callback (which is ... nested and leads to potential "callback hell"?). But stuck with two certain things:

• Extracting the value from the generated future with Redis connection
• Make the generated future so, that it won't block when it nested

So, actual code looks like this:

let server = socket.incoming().for_each(|(stream, addr)| {
    let engine_inner = self.engine.clone();
    let connections_inner = self.connections.clone();
    let handle_inner = handle.clone();

    let auth_middleware_callback = |request: &Request| {
        let handle_inner = handle.clone();
        let auth_middleware_inner = self.auth_middleware.clone();
        // Inside of this middleware we're checking and validating a specified token 
        let processing_result = auth_middleware_inner.borrow().process_request(request, &handle_inner);

        match processing_result {
            Ok(headers) => Ok(headers),
            Err(err) => {
                let formatted_error = format!("{}", err);
                let message = Cow::from(formatted_error);
                Err(TungsteniteError::Protocol(message))
            }
        }
    };

    accept_hdr_async(stream, auth_middleware_callback)
        // Process the messages
       .and_then(move |ws_stream| {
       // other code here...

The piece of work, that polling the Redis instance (yes, I know that is bad approach to get a value from the finished future (mostly it's for debugging proposes)):

fn get_user_id(&self, raw_token: String, handle: &Handle) -> Result<String> {
    let redis_socket_address = self.redis_address.parse().unwrap();
    let redis_connection = paired_connect(&redis_socket_address, handle);

    // Make the authentication before, if a password was specified.
    let get_user_id_future = redis_connection.and_then(move |connection| {
        // we will get here Ok("some user ID") if everything is good
        connection.send::<String>(resp_array!["GET", raw_token])
    });

    let mut processing_result;
    handle.spawn(get_user_id_future.then(move |response| {
        processing_result = match response {
            Ok(user_id) => Ok(String::from(user_id)),
            Err(_) => {
                let message = String::from("Token is expired or doesn't exist.");
                Err(PathfinderError::AuthenticationError(message))
            }
        };
        Ok(())
    }));

    processing_result
}

So, I have a couple of questions around this stuff:

• Does it possible to specify (or implement) a custom future, so that it will be used later in chain with the main part of handling of a client connection? I mean the case, which could be looks like this:

let auth_middleware_callback = |request: &Request| {
    let handle_inner = handle.clone();
    let auth_middleware_inner = self.auth_middleware.clone();
    let redis_future = auth_middleware_inner.borrow().process_request(request, &handle_inner);
    redis_future
};

accept_hdr_async(stream, auth_middleware_callback)
    // extract a value from redis_future
    .and_then((ws_stream, redis_future) {
        // ...
    })
    // Process the messages
    .and_then(move |ws_stream| {

Or in this case I should to go an another way (may be use specify token in the message, and process in inside or accept_async)?

• If return my code to a solution with accept_async, then can I return in the same time ws_stream and redis_future simultaneously? Or it possible so solve this issue somehow differently?
• Are there any suitable solutions for connection async libs (like redis-async, librdkafka, etc.) with tokio-tungstenite, so that they are working together in async way? Perhaps it sounds too generic, but nonetheless.

[daniel-abramov]

First of all, I did not quite get the idea of some of your questions (also I cannot understand how get_user_id() relates to tokio-tungstenite), but I'll try my best to answer your question.

If I got you right, you want to check the incoming user (upon reception of incoming websocket handshake request) and drop it if certain conditions for this particular user are not met, in your case you check the "condition" by doing some work around executing redis futures.

It is possible to implement what you want, but if your redis future depends on incoming WS-handshake header values, then there will be a portion of "blocking code" in your headers callback, i.e. you have to spawn/wait for a redis future to complete inside a callback (there are several ways to accomplish this, you may want to choose the one which better fits your needs).

As for returning a future from the headers callback -- currently it's not possible, we would need to update some tokio-tungstenite (and probably tungstenite) APIs to allow such chaining to support this behavior, but I got your point and it's BTW a good enhancement request -- returning future instead of a plain Result<> totally makes sense in tokio-tungstenite! Feel free to create a corresponding issue, so that we can mark it as an enhancement and consider its implementation. I don't plan to work on this particular enhancement in the near future (there are still some pending issues in tungstenite/tokio-tungstenite which have higher priority for us at the moment), but feel free to create a new pull request if you by chance can work on the implementation of this feature.

Hopefully I was able to answer your tungstenite related questions, so I would consider this issue as closed.