kayleg / cloud-pubsub Goto Github PK

https://cloud.google.com/sdk/gcloud/reference/beta/emulators/pubsub

To be usable as a standard error type, this type needs to implement std::error::Error, which implies both std::fmt::Debug and std::fmt::Display. Error can be trivially implemented, and Debug can be auto-derived, so this means providing a Display method.

Currently there should be a mounted file with credentials for the library to work. This is inconvenient in case the credentials are passed using an environment variable. I'm about to make a fork to fix this, but not sure the solution would be proper enough to contribute.

It might actually be beneficial to auto-refresh tokens by checking if they are close to expiration as having to spawn a separate thread for that seems laborious.

The current approach I've taken is as follows:

pub fn init_pubsub() -> cloud_pubsub::Client {
    ....

    let client = cloud_pubsub::Client::new_with_string(pubsub_credentials).unwrap();
    let refresh_client = client.clone();
    spawn(lazy(move |_ctx| {
        refresh_client.spawn_token_renew(GOOGLE_TOKEN_REFRESH_INTERVAL)
    }));
    client
}

But I can't imagine a scenario where one would not want to refresh token when it expires, nor do I think it's an implementation detail that the library user should care about.

What we could do is check token expiration on request and if it's close to be expired - spawn renew it right there and then.
It just has to be implemented in a performant way.

I do understand why the current approach has been taken: it makes the library easier to read and understand, but I'm sure we can come up with a solution that would work just as well without the extra burden on the user's side.

Hi ,

Trying to understand why you choose to use use std::sync::RwLock, instead of async lock like tokio::sync::RwLock.

The impact of sync lock on async program is very big . it's will hold system-thread , and block all other tasks that running on this thread.

change to async look will cause some api changes (sync functions will became async) , but if you ok with such change I will work on PR.

by looking at the payload i've noticed that we only publish the data to Pub/sub

#[derive(Deserialize, Clone, Serialize)]
pub struct EncodedMessage {
    data: String,
}

There's a really useful field named attributes that would be really nice to have. It helps filtering message within a subscription
(see https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage)

I'll make a PR of the forked version that I'm using that has this feature, maybe we can integrate it properly :)

Thanks!

Here's a nice example: https://github.com/hyperium/tonic/blob/master/examples/src/gcp/client.rs .

In subscription::get_messages the ack_ids are decoupled from the converted messages, since the function returns two vectors.

        let messages = response.received_messages.unwrap_or_default();
        let ack_ids: Vec<String> = messages
            .as_slice()
            .iter()
            .map(|packet| packet.ack_id.clone())
            .collect();
        let packets = messages
            .into_iter()
            .filter_map(|packet| match T::from(packet.message) {
                Ok(o) => Some(o),
                Err(e) => {
                    error!("Failed converting pubsub {}", e,);
                    None
                }
            })
            .collect();

        Ok((packets, ack_ids))

If messages cannot be converted, they are dropped. So how would I know which ack_id is for which message? On the happy path the are correlated by index, but if some messages are dropped, that is no longer the case.