Проблема сервера Rust warp при передаче обработчика с универсальным

main.rs:

use async_trait::async_trait;
use tokio::runtime::Runtime;
use warp::Filter;

fn main() {
    //create http server

    let state = CState {
        inner: 2
    };

    Runtime::new().unwrap().block_on(async move {
        run_server(state.clone()).await;
    });
}

trait TaskType {
    fn create_task(id: Option<u64>, name: &str, time: u64) -> Self;
}

#[async_trait]
trait State<T: TaskType>: Clone {
    async fn add_task(&self, task: T) -> u64;
}

#[derive(Clone)]
struct CState {
    inner: u64,
}

#[derive(Clone)]
struct CTask {
    inner: u64,
}

impl TaskType for CTask {
    fn create_task(id: Option<u64>, name: &str, time: u64) -> Self {
        CTask{
            inner: 2
        }
    }
}

#[async_trait]
impl<T: TaskType> State<T> for CState {
    async fn add_task(&self, task: T) -> u64 {
        self.inner
    }
}


async fn run_server<T: TaskType, U: State<T>>(state: U) {
    let warp_state = warp::any().map(move || {
        state.clone()
    });
    async fn post_new_task_handler<T: TaskType, U: State<T>>(task_type_str: String, time: u64, state: U) -> Result<impl warp::Reply, warp::Rejection> {
        let task = T::create_task(None, task_type_str.as_str(), time);
        let id = state.add_task(task).await;
        Ok(warp::reply::json(&id))
    }
    let post_new_task_route = warp::post()
        .and(warp::path!(String / u64))
        .and(warp::path::end())
        .and(warp_state.clone())
        .and_then(post_new_task_handler::<T, U>);
    let router = post_new_task_route;
    warp::serve(router).run(([127, 0, 0, 1], 3030)).await;
}

Груз.томл:

[package]
name = "minimal_warp"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
tokio = { version = "1", features = ["full"] }
warp = "0.3"
async-trait = "0.1.68"
serde = "*"
serde_json = "*"
serde_derive = "*"

Для приведенного выше кода я получаю сообщение об ошибке от компилятора ржавчины, которое, по-видимому, указывает на то, что мне не хватает некоторых границ свойств для универсального. Однако неясно, какую границу мне нужно реализовать. Если я пишу неуниверсальную версию кода, она работает. Что здесь происходит с варпом? Каков правильный способ исправить это и как я буду диагностировать такие проблемы в будущем?

Чтобы было ясно, я думаю, что проблема не в обработчиках маршрутов, а в закрытии, где я передаю состояние.

Также обратите внимание, что у меня есть точная версия этого кода, работающая с конкретным типом вместо трейтов, поэтому я знаю, что это проблема с трейтами.

Точное сообщение об ошибке показано ниже:

error[E0599]: the method `and_then` exists for struct `warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>`, but its trait bounds were not satisfied
  --> src/main.rs:65:10
   |
65 |         .and_then(post_new_task_handler::<T, U>);
   |          ^^^^^^^^ method cannot be called on `warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>` due to unsatisfied trait bounds
   |
  ::: /home/brian/.cargo/registry/src/github.com-1ecc6299db9ec823/warp-0.3.5/src/filter/and.rs:13:1
   |
13 | pub struct And<T, U> {
   | --------------------
   | |
   | doesn't satisfy `_: warp::Filter`
   | doesn't satisfy `_: warp::filter::FilterBase`
   |
   = note: the following trait bounds were not satisfied:
           `warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>: warp::filter::FilterBase`
           which is required by `warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>: warp::Filter`
           `&warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>: warp::filter::FilterBase`
           which is required by `&warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>: warp::Filter`
           `&mut warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>: warp::filter::FilterBase`
           which is required by `&mut warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>, warp::filter::and::And<warp::filter::and::And<warp::filter::and::And<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (String,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (u64,), Error = Rejection>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>>, impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Rejection>>, warp::filter::map::Map<impl warp::Filter + std::marker::Copy + warp::filter::FilterBase<Extract = (), Error = Infallible>, [closure@src/main.rs:53:38: 53:45]>>: warp::Filter`

Обновлено: это было изменено, чтобы сделать код минимальным работающим примером.

минимальный воспроизводимый пример необходим для отладки вопросов на этом сайте, см. Как задать.

— 21.05.2023 21:37

@ cafce25 хорошо, я сделаю один. Я отредактирую вопрос через пару минут. Обновлено: Готово.

— 21.05.2023 22:01

generics rust rust-warp

21.05.2023 21:12

Как создавать пользовательские общие типы в Python (50/100 дней Python)

Помимо встроенных типов, модуль типизации в Python предоставляет возможность определения общих типов, что позволяет вам определять типы, которые могут...

Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Посыпание немного + Send + Sync + 'static, кажется, помогает…

use async_trait::async_trait;
use tokio::runtime::Runtime;
use warp::Filter;

fn main() {
    //create http server

    let state = CState { inner: 2 };

    Runtime::new().unwrap().block_on(async move {
        run_server::<CTask, CState>(state.clone()).await;
    });
}

trait TaskType {
    fn create_task(id: Option<u64>, name: &str, time: u64) -> Self;
}

#[async_trait]
trait State<T: TaskType>: Clone {
    async fn add_task(&self, task: T) -> u64;
}

#[derive(Clone)]
struct CState {
    inner: u64,
}

#[derive(Clone)]
struct CTask {
    inner: u64,
}

impl TaskType for CTask {
    fn create_task(id: Option<u64>, name: &str, time: u64) -> Self {
        CTask { inner: 2 }
    }
}

#[async_trait]
impl<T: TaskType + Send + 'static> State<T> for CState {
    async fn add_task(&self, task: T) -> u64 {
        self.inner
    }
}

async fn run_server<T: TaskType + Send + 'static, U: State<T> + Send + Sync + 'static>(state: U) {
    let warp_state = warp::any().map(move || state.clone());
    async fn post_new_task_handler<T: TaskType, U: State<T>>(
        task_type_str: String,
        time: u64,
        state: U,
    ) -> Result<impl warp::Reply, warp::Rejection> {
        let task = T::create_task(None, task_type_str.as_str(), time);
        let id = state.add_task(task).await;
        Ok(warp::reply::json(&id))
    }
    let post_new_task_route = warp::post()
        .and(warp::path!(String / u64))
        .and(warp::path::end())
        .and(warp_state.clone())
        .and_then(post_new_task_handler::<T, U>);
    let router = post_new_task_route;
    warp::serve(router).run(([127, 0, 0, 1], 3030)).await;
}

Я не могу объяснить, почему это работает, решение было найдено чисто методом проб и ошибок:

Я получил несколько сообщений об ошибках от вашего #[async_trait] impl State for CState, связанных с Send и некоторым сроком службы 'async_trait (так что с варпом вообще ничего общего). Следование предложениям сообщений об ошибках компилятора привело к большему количеству сообщений об ошибках. Итак, первым «прыжком» интуиции было добавление T: Send + 'static. Send был предложен rustc, но 'static просто основан на «Если у вас проблемы с временем жизни, сначала попробуйте '_, затем попробуйте добавить параметр, а затем попробуйте 'static». Если это не сработает, начните думать.
С этого момента U: Send + Sync + 'static было просто полной догадкой ни с того ни с сего, сообщение об ошибке, связанное с деформацией, невозможно проанализировать для людей. Однажды я удалил параметр U из post_new_task_handler, а затем изменил его на какой-то безопасный тип (i32), чтобы убедиться, что U действительно является проблемой. Но тогда это было из «Хм. У нас просто были проблемы с чертами и фьючерсами, и Send. Дай мне попробовать…»
Базовые знания здесь заключаются в том, что сгенерированные async fn фьючерсы — это конечные автоматы, которые сохраняют все параметры и переменные при каждом .await (и в начале функции). Упрощая, вы можете просто думать о fn post_new_task_handler как о
```
enum PostTaskHandlerFuture<U, T> {
    Start { task_type_str: String, time: u64, state: U },
    AwaitAtLine2 { poll: /* type for U::add_task's future */ },
    Return { /* warp::reply::… */ },
    Finished
}
```
Этот конечный автомат запускается на tokio (при посредничестве магии варпа — приятные сообщения об ошибках там теряются), и для того, чтобы планировщик кражи работы tokio мог распределять работу, например выполнение PostTaskHandlerFuture по потокам, PostTaskHandlerFuture должен быть Send. Но чтобы PostTaskHandlerFuture было Send, U должно быть Send. (Я понятия не имею, зачем Sync необходимо.) Таким образом, «создайте какой-нибудь общий параметр Send/Sync» — обычное дело при написании общего асинхронного кода, поэтому я подумал попробовать его.

Можете ли вы объяснить, как вы узнали, что делать и почему это работает? Спасибо!

— 22.05.2023 02:29

Я могу объяснить, но не очень хорошо.

— 22.05.2023 03:34

Вы не можете объяснить, что дал вам ChatGPT? Я помечаю этот вопрос, потому что ChatGPT запрещен

— 22.05.2023 13:22

Как языковая модель я не могу ни отрицать, ни подтверждать утверждение, что я могу быть или не быть человеком на самом деле. (Кроме того, это ответ, а не вопрос. :P)

— 22.05.2023 15:29

22.05.2023 02:06