Pubsub implementation not usable in hyper handler

[bmoyet]

Hello,

I'm quite new to Rust, so I might be reporting something which is not an issue, forgive me if that's the case.

I'm trying to use the library to publish messages on Google Cloud PubSub, when some hyper endpoints are being called. But my project wouldn't build as soon as I'm calling the function I created to do so.

The build gives me the following error :

error[E0277]: `RefCell<Client<hyper_rustls::connector::HttpsConnector<HttpConnector>>>` cannot be shared between threads safely
  --> src/main.rs:27:25
   |
27 |     router.get("/test", Box::new(handler::test_handler));
   |                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `RefCell<Client<hyper_rustls::connector::HttpsConnector<HttpConnector>>>` cannot be shared between threads safely
   |
   = help: within `google_pubsub1::api::Pubsub<Client<hyper_rustls::connector::HttpsConnector<HttpConnector>>>`, the trait `Sync` is not implemented for `RefCell<Client<hyper_rustls::connector::HttpsConnector<HttpConnector>>>`
...
And a lot more message

From what I understand it is caused by the internal use of RefCell in this project which is not Sync whereas hyper needs it to be Sync.

To reproduce the issue I used the following repo https://github.com/zupzup/rust-minimal-web-service-hyper

My PubSub code is :

pub async fn send() -> Result<(), String> {
    let key = oauth2::read_service_account_key("xxx")
        .await
        .map_err(|err| format!("Cannot read service account credentials : {}", err))?;
    let secret = ServiceAccountAuthenticator::builder(key)
        .build()
        .await
        .map_err(|err| format!("Cannot build authenticator : {}", err))?;

    let hub = Pubsub::new(
        hyper::Client::builder()
            .build(hyper_rustls::HttpsConnector::with_native_roots()),
        secret);

    let msg = PubsubMessage { publish_time: None, message_id: None, attributes: None, data: Some(String::from("SGVsbG8gd29ybGQgIQ==")), ordering_key: None };
    let req = PublishRequest { messages: Some(vec![msg]) };

    let projects = hub.projects();
    println!("Sending new message");
    let result = projects
        .topics_publish(req, format!("projects/{}/topics/{}", "xxx", "xxx").as_str())
        .doit()
        .await;

    match result {
        Err(e) => match e {
            // The Error enum provides details about what exactly happened.
            // You can also just use its `Debug`, `Display` or `Error` traits
            Error::HttpError(_)
            |Error::Io(_)
            |Error::MissingAPIKey
            |Error::MissingToken(_)
            |Error::Cancelled
            |Error::UploadSizeLimitExceeded(_, _)
            |Error::Failure(_)
            |Error::BadRequest(_)
            |Error::FieldClash(_)
            |Error::JsonDecodeError(_, _) => println!("{}", e),
        },
        Ok(res) => println!("Success: {:?}", res),
    }

    return Ok(());
}

And I'm modifying the rust-minimal-web-service-hyper project this way in handler.rs :

pub async fn test_handler(ctx: Context) -> String {
    let res = pubsub::send()
        .await
        .map(|_| "Ok".to_string())
        .unwrap_or("Error".to_string());

    format!("test called, state_thing was: {}. Result : {}", ctx.state.state_thing, res)
}

When calling my method from a main function it works correctly, but as soon as I'm using it inside a hyper handler the build is failing

[Byron]

Indeed that's more of a Rust question, let's see if this can be sorted. Also I am happy the build is failing, which is one of the major benefits when using Rust (AKA fearless concurrency :D).

router.get("/test", ENDPOINT) wants to use ENDPOINT in a thread as part of its internal machinery and do so by reference, requiring the Future to be Sync. The RefCell used in its belly is not Sync though and can't be accessed by multiple threads safely without a race condition: after all it wants to assure that using the same thing mutably panics at runtime.

One way to solve this would be to make the APIs threadsafe, so instead of using a RefCell, one could use a Mutex. It's something I am happy to do, but maybe another option could be explored first.

Is it possible to do something like this?

let sender = std::sync::Mutex::new(pubsub::send());
let fut_guard = sender.lock();
fut_guard.await…

I could imagine that a future within a mutex guard can't be awaited anymore, but it would be nice to try just once. If that doesn't work, I would have to patch the crate you are using which is certainly desirable.

[Byron]

I took a closer look just to realize that the RefCell pieces were leftovers of a once synchonous, blocking API. Now everything in hyper and yup-oauth2 is async and naturally supports `Sync.

This allowed a lot of simplification, leading to pubsub1 being republished as v2.0.3. It should have none of the issue described here.

All other APIs are on their way of being republished.

Thanks again for bringing this up - I think it let to a major improvement of all the GoogleAPI crates.

[bmoyet]

Thank you for this quick fix, appreciate it.

I tried the new version but it seems that there is another issue when I try my setup above. I now get a missing Send error on ProjectTopicPublishCall.

Full error below :

error: future cannot be sent between threads safely
  --> src/main.rs:27:25
   |
27 |     router.get("/test", Box::new(handler::test_handler));
   |                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ future returned by `test_handler` is not `Send`
   |
   = help: the trait `std::marker::Send` is not implemented for `dyn Delegate`
note: future is not `Send` as this value is used across an await
  --> src/handler.rs:30:18
   |
30 |        let result = hub.projects()
   |   __________________^
   |  |__________________|
   | ||
31 | ||         .topics_publish(req, format!("projects/{}/topics/{}", "nice-pen-498", "local-configuration-flush-event").as_str())
   | ||__________________________________________________________________________________________________________________________- has type `ProjectTopicPublishCall<'_>` which is not `Send`
32 | |          .doit()
33 | |          .await;
   | |               ^- `hub.projects()
        .topics_publish(req, format!("projects/{}/topics/{}", "nice-pen-498", "local-configuration-flush-event").as_str())` is later dropped here
   | |_______________|
   |                 await occurs here, with `hub.projects()
        .topics_publish(req, format!("projects/{}/topics/{}", "nice-pen-498", "local-configuration-flush-event").as_str())` maybe used later
   = note: required for the cast to the object type `dyn Handler`

Is this the expected behavior ?

[Byron]

Thanks! We will get there :). What about now with v2.0.4?

[bmoyet]

Your last correction make it work, it seems good for now.

Thanks again for your reactivity 👍

[Byron]

Thank you too! Let me republish everything, again :D - I was a bit too confident previously 🤦‍♂️.