Rusoto: Composing rusoto futures

[softprops]

Composing and tranaforming rusoto futures atm can be slightly confusing as result loses rusoto future methods like sync and timeout. Because rusoto clients hide the thier tokio reactor core is the recommendation to call wait instead? If that is possible why does sync exist?

[srijs]

The sync method is really only meant to be used in an environment that doesn't use futures otherwise and want to block on the result. If your application already uses futures and tokio and you want to block on the result of the future, I'd recommend you use .wait() in a thread outside of the tokio threadpool. I'll make sure to update the docs to clarify this!

With regards to timeouts, can you go into more detail what you would expect .set_timeout to do on a future that is composed from multiple rusoto calls?

[softprops]

I'd like see timeout would work the same on the aggregate. I'm coming from a Scala background and am used to calling a method that waits with a timeout at the end of a call chain https://www.scala-lang.org/api/current/scala/concurrent/Await$.html

[xrl]

@srijs how should I use rusoto without using wait()? I try something like this and I get issues with rusoto client not being send:

extern crate rusoto_core;
extern crate rusoto_s3;
#[macro_use] extern crate futures;
extern crate tokio;
//#[macro_use] extern crate log;
extern crate env_logger;

use futures::Future;

fn main() {
    env_logger::init();

    tokio::run(futures::lazy(|| {
        let client = rusoto_s3::S3Client::simple(rusoto_core::region::Region::UsWest2);

        let list = rusoto_s3::S3::list_buckets(&client).then(|x| {
            println!("x: {:?}", x);
            Ok::<(),()>(())
        });

        list.then(|res| {
            println!("res: {:?}", res);
            Ok(())
        })
    }));
}

error[E0277]: `(dyn rusoto_core::client_inner::ClientInnerFuture<Error=rusoto_s3::ListBucketsError, Item=rusoto_s3::ListBucketsOutput> + 'static)` cannot be sent between threads safely
  --> src/main.rs:13:5
   |
13 |     tokio::run(futures::lazy(|| {
   |     ^^^^^^^^^^ `(dyn rusoto_core::client_inner::ClientInnerFuture<Error=rusoto_s3::ListBucketsError, Item=rusoto_s3::ListBucketsOutput> + 'static)` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `(dyn rusoto_core::client_inner::ClientInnerFuture<Error=rusoto_s3::ListBucketsError, Item=rusoto_s3::ListBucketsOutput> + 'static)`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<(dyn rusoto_core::client_inner::ClientInnerFuture<Error=rusoto_s3::ListBucketsError, Item=rusoto_s3::ListBucketsOutput> + 'static)>`
   = note: required because it appears within the type `std::boxed::Box<(dyn rusoto_core::client_inner::ClientInnerFuture<Error=rusoto_s3::ListBucketsError, Item=rusoto_s3::ListBucketsOutput> + 'static)>`
   = note: required because it appears within the type `rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>`
   = note: required because it appears within the type `futures::future::chain::Chain<rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>, futures::FutureResult<(), ()>, [closure@src/main.rs:16:62: 19:10]>`
   = note: required because it appears within the type `futures::Then<rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>, std::result::Result<(), ()>, [closure@src/main.rs:16:62: 19:10]>`
   = note: required because it appears within the type `futures::future::chain::Chain<futures::Then<rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>, std::result::Result<(), ()>, [closure@src/main.rs:16:62: 19:10]>, futures::FutureResult<(), ()>, [closure@src/main.rs:20:19: 23:10]>`
   = note: required because it appears within the type `futures::Then<futures::Then<rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>, std::result::Result<(), ()>, [closure@src/main.rs:16:62: 19:10]>, std::result::Result<(), ()>, [closure@src/main.rs:20:19: 23:10]>`
   = note: required because it appears within the type `futures::future::lazy::_Lazy<[closure@src/main.rs:13:30: 24:6], futures::Then<futures::Then<rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>, std::result::Result<(), ()>, [closure@src/main.rs:16:62: 19:10]>, std::result::Result<(), ()>, [closure@src/main.rs:20:19: 23:10]>>`
   = note: required because it appears within the type `futures::Lazy<[closure@src/main.rs:13:30: 24:6], futures::Then<futures::Then<rusoto_core::RusotoFuture<rusoto_s3::ListBucketsOutput, rusoto_s3::ListBucketsError>, std::result::Result<(), ()>, [closure@src/main.rs:16:62: 19:10]>, std::result::Result<(), ()>, [closure@src/main.rs:20:19: 23:10]>>`
   = note: required by `tokio::run`

[srijs]

@xrl The latest rusoto release predates the "new" tokio runtime, which is why it's not straight-forward to use with it. The next release will fix this and have full compatibility, including RusotoFuture being send. For now my recommendation would be to either use tokio-core instead as a runtime (which can be used with both old and "new" tokio libraries), or alternatively use the tokio-current-thread crate (however this requires additional set-up for things like timers).

@softprops Setting a timeout on the aggregate feels like a concern that's not necessarily rusoto related? tokio generally offers timeout functionality that I'd use for cases like this.

[softprops]

@srijs all of your answers made sense to me. Some async examples may be helpful for others to follow along with under an /examples directory. Feel free to close out this issue. 👍

[srijs]

Cool, I've opened #1086 to track your suggestion!