error[E0277]: the trait bound `impl Future<Output = Result<impl Reply, Rejection>>: Reply` is not satisfied

#![deny(warnings)]
use sqlx::postgres::PgPoolOptions;
use warp::{reject::Reject, Filter};
use rusoto_core::{Region, credential::StaticProvider};
use rusoto_s3::{S3, S3Client, ListObjectsV2Request, DeleteObjectsRequest, Delete, ObjectIdentifier, DeleteObjectsError};


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let database_url = ..../
    let pool = PgPoolOptions::new() 
        .max_connections(5)
        .connect(&database_url)
        .await?;

    // 配置 OSS 访问参数
      ..../

    // 创建 S3 客户端
    let client = S3Client::new_with(
        rusoto_core::request::HttpClient::new().expect("Failed to create HTTP client"),
        StaticProvider::new_minimal(access_key_id.to_string(), access_key_secret.to_string()),
        Region::Custom { name: region_id.to_owned(), endpoint: endpoint.to_owned() },
    );
    let routes = warp::any()
    .map( || cleanup_unreachable_data(&client, &pool));
    warp::serve(routes).run((....)).await;

    Ok(())
}

#[derive(Debug)]
struct SqlxErrorRejection(sqlx::Error);
impl Reject for SqlxErrorRejection{}


async fn cleanup_unreachable_data(client: &S3Client, pool: &sqlx::PgPool)
 ->  std::result::Result<impl warp::reply::Reply, warp::Rejection> {
    let db_paths: Result<Vec<String>, sqlx::Error> = sqlx::query!(
    "SELECT DISTINCT CONCAT(merchant_id, '/', logo) as path FROM store_info")
        .fetch_all(pool)
        .await
        .map_err(SqlxErrorRejection)?
        .iter()
        .map(|row| {
            row.path.clone().ok_or(sqlx::Error::Decode(Box::new(sqlx::error::Error::ColumnNotFound("path".to_owned()))))
        })
        .collect();
        println!("{:?}",db_paths);  
    // 检查每条记录是否在数据库中的 OSS 路径列表中

    let request = ListObjectsV2Request {
        ..Default::default()
    };
    let response = &client.list_objects_v2(request).await.unwrap();
    
    let mut all_object_paths: Vec<String> = Vec::new();

    if let Some(contents) = &response.contents {
        for object in contents {
            let object_path = object.key.clone().unwrap_or_default();
            println!("Object Path: {}", object_path);
            all_object_paths.push(object_path);
            }
        }
        if let Ok(record) = &db_paths {
            // let record_path = format!("{}/{}", record.merchant_id, record.logo);
            println!("Record Path: {:?}", record);
            for path in record {
                println!("Record Path: {}", path);
                all_object_paths.retain(|object_path| object_path != path);
            } 
            println!("{:?}",all_object_paths);
            // 如果数据库中的记录不在 OSS 路径列表中,执行删除
            for object_path in &all_object_paths {
                delete_record_from_object_storage(&client, &object_path).await?;
            }
            
}
     Ok(warp::http::StatusCode::OK)
}

#[derive(Debug)]
struct RusotoDeleteObjectsError(pub rusoto_core::RusotoError<DeleteObjectsError>);

// 实现 Reject trait
impl Reject for RusotoDeleteObjectsError {}

async fn delete_record_from_object_storage(
        client: &S3Client,
        path: &str,
    ) ->  std::result::Result<impl warp::reply::Reply, warp::Rejection> {
        let delete_request = DeleteObjectsRequest {
            delete: Delete{
                objects: vec![ObjectIdentifier {
                    key: path.to_string(),
                    ..Default::default()
                }],
                ..Default::default()
            },
            ..Default::default()
        };
        client.delete_objects(delete_request).await
        .map_err(|err| RusotoDeleteObjectsError(err.into()));
    
        println!("Deleting Object: {}", path);  
            
        Ok(warp::http::StatusCode::OK)
}

Error

error[E0277]: the trait bound `impl Future<Output = Result<impl Reply, Rejection>>: Reply` is not satisfied
  --> src\main.rs:30:17
   |
30 |     warp::serve(routes).run(([0, 0, 0, 0], 9000)).await;
   |     ----------- ^^^^^^ the trait `Reply` is not implemented for `impl Future<Output = Result<impl Reply, Rejection>>`
   |     |
   |     required by a bound introduced by this call
   |
   = help: the following other types implement trait `Reply`:
             Box<T>
             Cow<'static, str>
             log::internal::Logged
             Response<T>
             trace::internal::Traced
             warp::reply::Json
             Html<T>
             warp::reply::sealed::Reply_
           and 14 others
   = note: required for `(impl Future<Output = Result<impl Reply, Rejection>>,)` to implement `Reply`
note: required by a bound in `serve`
  --> C:\Users\DSpc\.cargo\registry\src\rsproxy.cn-8f6827c7555bfaf8\warp-0.3.2\src\server.rs:26:17
   |
23 | pub fn serve<F>(filter: F) -> Server<F>
   |        ----- required by a bound in this function
...
26 |     F::Extract: Reply,
   |                 ^^^^^ required by this bound in `serve`

error[E0277]: the trait bound `impl Future<Output = Result<impl Reply, Rejection>>: Reply` is not satisfied
   --> src\main.rs:30:25
    |
30  |     warp::serve(routes).run(([0, 0, 0, 0], 9000)).await;
    |                         ^^^ the trait `Reply` is not implemented for `impl Future<Output = Result<impl Reply, Rejection>>`
    |
    = help: the following other types implement trait `Reply`:
              Box<T>
              Cow<'static, str>
              log::internal::Logged
              Response<T>
              trace::internal::Traced
              warp::reply::Json
              Html<T>
              warp::reply::sealed::Reply_
            and 14 others
    = note: required for `(impl Future<Output = Result<impl Reply, Rejection>>,)` to implement `Reply`
note: required by a bound in `warp::Server::<F>::run`
   --> C:\Users\DSpc\.cargo\registry\src\rsproxy.cn-8f6827c7555bfaf8\warp-0.3.2\src\server.rs:127:35
    |
127 |     <F::Future as TryFuture>::Ok: Reply,
    |                                   ^^^^^ required by this bound in `Server::<F>::run`
...
131 |     pub async fn run(self, addr: impl Into<SocketAddr>) {
    |                  --- required by a bound in this associated function

For more information about this error, try `rustc --explain E0277`.
error: could not compile `server` (bin "server") due to 2 previous errors

Not a warp user. But the error says routes is not of type implemented with Reply, and it seems you're mistakenly using methods on Filter in warp - Rust

// This takes a normal function
fn map<F>(self, fun: F) -> Map<Self, F>
where
    Self: Sized,
    F: Func<Self::Extract> + Clone,

// This takes a normal function that returns 
// a TryFuture (i.e. a Future whose output is a Result)
// as `F: Func` and `F::Output: TryFuture` say
fn and_then<F>(self, fun: F) -> AndThen<Self, F>
where
    Self: Sized,
    F: Func<Self::Extract> + Clone,
    F::Output: TryFuture + Send,
    <F::Output as TryFuture>::Error: CombineRejection<Self::Error>,

rustexplorer: minimize your code

I have corrected the error and it works if I replace map with and_then
let routes = warp::any()
.and_then( || cleanup_unreachable_data());

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.