Блог

Я пытаюсь заставить асинхронное закрытие работать в and_then фильтре Warp .

Это самый маленький пример, который я мог придумать, где я достаточно уверен, что не упустил никаких важных деталей:

 use std::{convert::Infallible, sync::Arc, thread, time};
use tokio::sync::RwLock;
use warp::Filter;

fn main() {
    let man = Manifest::new();

    let check = warp::path("updates").and_then(|| async move { GetAvailableBinaries(amp;man).await });
}

async fn GetAvailableBinaries(man: amp;Manifest) -> Result<impl warp::Reply, Infallible> {
    Ok(warp::reply::json(amp;man.GetAvailableBinaries().await))
}

pub struct Manifest {
    binaries: Arc<RwLock<Vec<i32>>>,
}

impl Manifest {
    pub fn new() -> Manifest {
        let bins = Arc::new(RwLock::new(Vec::new()));

        thread::spawn(move || async move {
            loop {
                thread::sleep(time::Duration::from_millis(10000));
            }
        });

        Manifest { binaries: bins }
    }

    pub async fn GetAvailableBinaries(amp;self) -> Vec<i32> {
        self.binaries.read().await.to_vec()
    }
}
  

Я использую:

 [dependencies]
tokio = { version = "0.2", features = ["full"] }
warp = { version = "0.2", features = ["tls"] }
  

Ошибка:

 error[E0525]: expected a closure that implements the `Fn` trait, but this closure only implements `FnOnce`
 --> src/main.rs:9:48
  |
9 |     let check = warp::path("updates").and_then(|| async move { GetAvailableBinaries(amp;man).await });
  |                                       -------- ^^^^^^^^^^^^^ ------------------------------------ closure is `FnOnce` because it moves the variable `man` out of its environment
  |                                       |        |
  |                                       |        this closure implements `FnOnce`, not `Fn`
  |                                       the requirement to implement `Fn` derives from here
  

После создания Manifest реализации Clone вы можете исправить ошибку, выполнив балансировку при клонировании объекта манифеста:

 fn main() {
    let man = Manifest::new();

    let check = warp::path("updates").and_then(move || {
        let man = man.clone();
        async move { get_available_binaries(amp;man).await }
    });

    warp::serve(check);
}
  

Это перемещается man в переданное закрытие and_then , а затем предоставляет клон man асинхронному блоку при каждом выполнении закрытия. Затем асинхронный блок владеет этими данными и может ссылаться на них, не беспокоясь о выполнении future после освобождения данных.

Я не уверен, что это то, к чему вы стремитесь, но это решение подходит для меня:

 use std::{convert::Infallible, sync::Arc, thread, time};
use tokio::sync::RwLock;
use warp::Filter;

fn main() {
    let man = Manifest::new();

    let check = warp::path("updates").and_then(|| async { GetAvailableBinaries(amp;man).await });
}

async fn GetAvailableBinaries(man: amp;Manifest) -> Result<impl warp::Reply, Infallible> {
    Ok(warp::reply::json(amp;man.GetAvailableBinaries().await))
}

#[derive(Clone)]
pub struct Manifest {
    binaries: Arc<RwLock<Vec<i32>>>,
}

impl Manifest {
    pub fn new() -> Manifest {
        let bins = Arc::new(RwLock::new(Vec::new()));

        thread::spawn(move || async {
            loop {
                thread::sleep(time::Duration::from_millis(10000));
                //mutate bins here
            }
        });

        Manifest { binaries: bins }
    }

    pub async fn GetAvailableBinaries(amp;self) -> Vec<i32> {
        self.binaries.read().await.to_vec()
    }
}
  

move Вот причина, по которой компилятор выдал предупреждение относительно подписи : let check = warp::path("updates").and_then(|| async move { GetAvailableBinaries(amp;man).await }); . Это означает, что все, на что ссылается это закрытие, будет перенесено в контекст закрытия. В этом случае компилятор не может гарантировать, что закрытие будет Fn выполнено, а FnOnce означает, что закрытие может быть гарантировано только один раз.