Rusoto: how to make sync calls

help
1

Hello all,

First of all, thanks a lot for all the amazing job put into those libraries!!

I'm trying to integrate rusoto_s3 and rusoto_sqs (both versions 0.43.0) in a project that was not using anything async at first. I'd like to keep it that way if possible.

From the official guide, there is a function to turn rusoto futures into blocking calls:

Rusoto supports returning Futures to be executed later as well as synchronous, blocking calls. The .sync() function on Rusoto Futures allows this behavior.

Under the hood it uses a tokio::runtime to immediately run the Future..

From: https://rusoto.org/futures.html

But I can't make it work :cry: I feel like I'm failing to import the right crate / trait, but I don't know what.

Here is my current attempt:

use rusoto_core::Region;
use rusoto_s3::{GetObjectRequest, S3Client, S3};
let client = S3Client::new(Region::default());

let x = client
    .get_object(GetObjectRequest {
        bucket: "yadada".to_owned(),
        key: "yadada".to_owned(),
        ..Default::default()
    })
    .sync();

and the error I receive:

error[E0599]: no method named `sync` found for struct `std::pin::Pin<std::boxed::Box<dyn core::future::future::Future<Output = std::result::Result<rusoto_s3::generated::GetObjectOutput, rusoto_core::error::RusotoError<rusoto_s3::generated::GetObjectError>>> + std::marker::Send>>` in the current scope
  --> event-utils/src/s3.rs:74:14
   |
74 |             .sync();
   |              ^^^^ method not found in `std::pin::Pin<std::boxed::Box<dyn core::future::future::Future<Output = std::result::Result<rusoto_s3::generated::GetObjectOutput, rusoto_core::error::RusotoError<rusoto_s3::generated::GetObjectError>>> + std::marker::Send>>`

Can any one point me into the right direction? Maybe the guide is a little out-of-date?

2

I don't know about the sync function you mention, but you can create a Tokio Runtime object and use its block_on function to wait for your future.

let x_fut = client
    .get_object(GetObjectRequest {
        bucket: "yadada".to_owned(),
        key: "yadada".to_owned(),
        ..Default::default()
    });

let x = runtime.block_on(x_fut);
3

It looks like the RusotoFuture type and its sync() method don't exists an more. Plugging them it into the search for rusoto_core give no useful results. You may want to create an issue against the rusoto repo (or even a PR!) to update the docs.

As @alice mentioned, tokio's block_on() is normally the best way to translate from the async world to the sync one.

1 Like
4

Nice, interesting solution, thanks @alice and @Michael-F-Bryan :smiley:

My previous understanding was that block_on() was meant to be used only once, from main().

Do you know if it would be ok to create a basic scheduler runtime each time?

let basic_rt = tokio::runtime::Builder::new()
    .basic_scheduler()
    .build()?;

let x_fut = client
    .get_object(GetObjectRequest {
        bucket: "yadada".to_owned(),
        key: "yadada".to_owned(),
        ..Default::default()
    });

let x = basic_rt.block_on(x_fut);

I guess I could share the same runtime, but since the rusoto calls are far apart in the code, I'd rather avoid passing it as parameter in multiple places.

You may want to create an issue against the rusoto repo (or even a PR!) to update the docs.

Sure, I'll open an issue offering to update that part of the guide.

5

Creating a runtime for every call is rather expensive, even if it's only the basic one. Also, you need to call the enable_all method to be able to use Tokio's IO primitives in that runtime. Can't you share the runtime together with the client?

6

In fact yes, I can carry the runtime as part a struct :wink:
That way I'll have the tokio runtime and S3Client together, and I will be able to implement a simple synchronous API for the rest of my code.

I'll share here in case someone else finds it useful:

struct SyncS3 {
    client: S3Client,
    runtime: Runtime,
}

impl SyncS3 {
    fn new() -> Result<Self, Error> {
        Ok(SyncS3 {
            client: S3Client::new(Region::default()),
            runtime: Builder::new().basic_scheduler().enable_all().build()?,
        })
    }

    fn get_object(&mut self, request: GetObjectRequest) -> Result<String, Error> {
        let response = self.runtime.block_on(self.client.get_object(request))?;
        let mut body = String::new();
        response
            .body
            .unwrap()
            .into_blocking_read()
            .read_to_string(&mut body)?;
        Ok(body)
    }
}
2 Likes
7

Nice solution. Obviously you lose the nice things that come with async (e.g. concurrency), but as an effort to wrap an async API so it can be used by synchronous code it looks pretty good :+1:

Would it make things more readable to merge the object retrieval and response reading into a single future?

use tokio::io::AsyncReadExt;

impl SyncS3 {
    ...
    fn get_object(&mut self, request: GetObjectRequest) -> Result<String, Error> {
        self.runtime.block_on(async move {
            let response = self.client.get_object(request).await?;

            let mut body = String::new();
            response.body
                .expect("The response should have a body because we haven't read it yet")
                .into_async_read()
                .read_to_string(&mut body).await?;

            Ok(body)
        })
    }
}
8

Yes, indeed :+1: I haven't thought about using the async move block there

1 Like
9

This topic was automatically closed 90 days after the last reply. New replies are no longer allowed.