Warp: and_then() cant decipher which function args to use

[hamptokr]

I am using warp in a fairly large private project and can't post that code here, but I am able to reproduce the same error in a slightly tweaked version of the todos example. The only additions are adding a users_index, creating the create_user handler, and then adding it to the api with an additional or(create_user) call.

The error given is that for some reason the create_user handler expects the function arguments of the create handler (the one that expects a Todo and a Db). Been testing things for a few hours and can't pinpoint the issue, but I can tell you order does matter. If you put create_user before create. Then the error is that create expects a function with no args.

Example code

#![deny(warnings)]

use std::env;
use std::sync::Arc;
use tokio::sync::Mutex;
use warp::{http::StatusCode, Filter};
use serde_derive::{Deserialize, Serialize};

/// So we don't have to tackle how different database work, we'll just use
/// a simple in-memory DB, a vector synchronized by a mutex.
type Db = Arc<Mutex<Vec<Todo>>>;

#[derive(Debug, Deserialize, Serialize, Clone)]
struct Todo {
    id: u64,
    text: String,
    completed: bool,
}

/// Provides a RESTful web server managing some Todos.
///
/// API will be:
///
/// - `GET /todos`: return a JSON list of Todos.
/// - `POST /todos`: create a new Todo.
/// - `PUT /todos/:id`: update a specific Todo.
/// - `DELETE /todos/:id`: delete a specific Todo.
#[tokio::main]
async fn main() {
    if env::var_os("RUST_LOG").is_none() {
        // Set `RUST_LOG=todos=debug` to see debug logs,
        // this only shows access logs.
        env::set_var("RUST_LOG", "todos=info");
    }
    pretty_env_logger::init();

    // These are some `Filter`s that several of the endpoints share,
    // so we'll define them here and reuse them below...

    // Turn our "state", our db, into a Filter so we can combine it
    // easily with others...
    let db = Arc::new(Mutex::new(Vec::<Todo>::new()));
    let db = warp::any().map(move || db.clone());

    // Just the path segment "todos"...
    let todos = warp::path("todos");

    // Combined with `end`, this means nothing comes after "todos".
    // So, for example: `GET /todos`, but not `GET /todos/32`.
    let todos_index = todos.and(warp::path::end());
    let users_index = warp::path("users").and(warp::path::end());

    // Combined with an id path parameter, for refering to a specific Todo.
    // For example, `POST /todos/32`, but not `POST /todos/32/something-more`.
    let todos_id = todos.and(warp::path::param::<u64>()).and(warp::path::end());

    // When accepting a body, we want a JSON body
    // (and to reject huge payloads)...
    let json_body = warp::body::content_length_limit(1024 * 16).and(warp::body::json());

    // For `GET /todos` also allow optional query parameters to allow for paging of Todos.
    let list_options = warp::query::<ListOptions>();

    // We'll make one of our endpoints admin-only to show how authentication filters are used
    let admin_only = warp::header::exact("authorization", "Bearer admin");

    // Next, we'll define each our 4 endpoints:

    // `GET /todos`
    let list = warp::get()
        .and(todos_index)
        .and(list_options)
        .and(db.clone())
        .and_then(list_todos);

    // `POST /todos`
    let create = warp::post()
        .and(todos_index)
        .and(json_body)
        .and(db.clone())
        .and_then(create_todo);

    let create_user = warp::post()
        .and(users_index)
        .and(json_body)
        .and(db.clone())
        .and_then(move || {});

    // `PUT /todos/:id`
    let update = warp::put()
        .and(todos_id)
        .and(json_body)
        .and(db.clone())
        .and_then(update_todo);

    // `DELETE /todos/:id`
    let delete = warp::delete()
        .and(todos_id)
        // It is important to put the auth check _after_ the path filters.
        // If we put the auth check before, the request `PUT /todos/invalid-string`
        // would try this filter and reject because the authorization header doesn't match,
        // rather because the param is wrong for that other path.
        .and(admin_only.clone())
        .and(db.clone())
        .and_then(delete_todo);

    // Combine our endpoints, since we want requests to match any of them:
    let api = list.or(create).or(create_user).or(update).or(delete);

    // View access logs by setting `RUST_LOG=todos`.
    let routes = api.with(warp::log("todos"));

    // Start up the server...
    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

// These are our API handlers, the ends of each filter chain.
// Notice how thanks to using `Filter::and`, we can define a function
// with the exact arguments we'd expect from each filter in the chain.
// No tuples are needed, it's auto flattened for the functions.

// The query parameters for list_todos.
#[derive(Debug, Deserialize)]
struct ListOptions {
    offset: Option<usize>,
    limit: Option<usize>,
}

/// GET /todos?offset=3&limit=5
async fn list_todos(opts: ListOptions, db: Db) -> Result<impl warp::Reply, warp::Rejection> {
    // Just return a JSON array of todos, applying the limit and offset.
    let todos = db.lock().await;
    let todos: Vec<Todo> = todos
        .clone()
        .into_iter()
        .skip(opts.offset.unwrap_or(0))
        .take(opts.limit.unwrap_or(std::usize::MAX))
        .collect();
    Ok(warp::reply::json(&todos))
}

/// POST /todos with JSON body
async fn create_todo(create: Todo, db: Db) -> Result<impl warp::Reply, warp::Rejection> {
    log::debug!("create_todo: {:?}", create);

    let mut vec = db.lock().await;

    for todo in vec.iter() {
        if todo.id == create.id {
            log::debug!("    -> id already exists: {}", create.id);
            // Todo with id already exists, return `400 BadRequest`.
            return Ok(StatusCode::BAD_REQUEST);
        }
    }

    // No existing Todo with id, so insert and return `201 Created`.
    vec.push(create);

    Ok(StatusCode::CREATED)
}

/// PUT /todos/:id with JSON body
async fn update_todo(id: u64, update: Todo, db: Db) -> Result<impl warp::Reply, warp::Rejection> {
    log::debug!("update_todo: id={}, todo={:?}", id, update);
    let mut vec = db.lock().await;

    // Look for the specified Todo...
    for todo in vec.iter_mut() {
        if todo.id == id {
            *todo = update;
            return Ok(warp::reply());
        }
    }

    log::debug!("    -> todo id not found!");

    // If the for loop didn't return OK, then the ID doesn't exist...
    Err(warp::reject::not_found())
}

/// DELETE /todos/:id
async fn delete_todo(id: u64, db: Db) -> Result<impl warp::Reply, warp::Rejection> {
    log::debug!("delete_todo: id={}", id);

    let mut vec = db.lock().await;

    let len = vec.len();
    vec.retain(|todo| {
        // Retain all Todos that aren't this id...
        // In other words, remove all that *are* this id...
        todo.id != id
    });

    // If the vec is smaller, we found and deleted a Todo!
    let deleted = vec.len() != len;

    if deleted {
        // respond with a `204 No Content`, which means successful,
        // yet no body expected...
        Ok(StatusCode::NO_CONTENT)
    } else {
        log::debug!("    -> todo id not found!");
        // Reject this request with a `404 Not Found`...
        Err(warp::reject::not_found())
    }
}

[seanmonstar]

This is where I'm to look?

    let create_user = warp::post()
        .and(users_index)
        .and(json_body)
        .and(db.clone())
        .and_then(move || {});

The function passed to and_then in that example will need the JSON and DB arguments to work. Right? Or are you saying that the compiler message isn't clear? That's also true, due to trait hackery needed for Func to work.

[hamptokr]

Well even if you had a User struct and passed that:

let create_user = warp::post()
        .and(users_index)
        .and(json_body)
        .and(db.clone())
        .and_then(move |u: User, db: Db| {});

You will get a compiler error saying and_then expected args Func<(Todo, Db)>, and not Func<(User, Db)>

[seanmonstar]

Ohh, I see what you mean. Yea, that's coming from the inference applied to warp::body::json::<T>().

[hamptokr]

Ahhh that fixed it! Thanks!!