yoshidan / google-cloud-rust Goto Github PK

What is the correct way to setup client config to get it working with workload identity from GCP?

I am using a patched dependency with the main branch since crates are updated manually. Using default auth does not work like it does in local.

ClientConfig::default().with_auth().await.unwrap()

results in thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: HttpError(reqwest::Error { kind: Decode, source: Error("expected value", line: 1, column: 1) })', /root/project/src/router.rs:22:51

I am currently doing some load testing comparing my previous app, built in python, and this new application built in rust using your google-cloud-rust repo and with the rust application, Im getting the below error when sending a high volume of requests to pubsub.
thread 'actix-rt|system:0|arbiter:0' panicked at 'calledResult::unwrap()on anErrvalue: GAX(Auth(HttpError(reqwest::Error { kind: Request, url: Url { scheme: "http", cannot_be_a_base: false, username: "", password: None, host: Some(Ipv4(169.xxx.xxx.xxx)), port: None, path: "/computeMetadata/v1/instance/service-accounts/default/token", query: Some("scopes%3Dhttps%3A%2F%2Fwww.googleapis.com%2Fauth%2Fcloud-platform%2Chttps%3A%2F%2Fwww.googleapis.com%2Fauth%2Fpubsub.data"), fragment: None }, source: TimedOut })))'

the same number of requests are sent to the python app, which is forwarding the request to the same pubsub service in GCP, and I do not see any failures/errors on the python app.

Any chance you could help me understand why this is happening and how I can fix it.

Cheers

Hi, please I'm trying to copy files from source to destination. I observed the API isn't included. Is there a reason why it's omitted? Can I make a pull request to add the copy API?

Hi, I'm trying to use with_credentials, but it appears not to be in the latest version of the crate although it is in the main branch. Also, importing the git repo as a dependency doesn't seem to work (maybe because the repo is a workspace). Please what can I do?

At work for high performance we have apis that operates in bulk not in single message.
Maybe adding a new method in publisher that support an iterable of ReservedMessages could be useful to prevent send one by one message to the channel. Maybe an enum Reserved::Multi and Reserved::Single could be useful to detect if is single or not for the matching here.

First of all, thanks a lot for all the work on this crate. It looks very promising ! 🥇

It seems the current google_cloud_storage::client::Clientapi does not allow Range
requests for now.

Google Cloud Storage documentation (https://cloud.google.com/storage/docs/xml-api/get-object-download) mentions:

GET requests for objects can include a Range header as defined in the HTTP 1.1 RFC to limit the scope of the returned data within the object

It would be nice if there was a way to supply a range header to the download_object and download_streamed_object http calls.

Pubsub supports a subscription.seek method to seek subscriptions back to a point in time or a saved snapshot. Can google-cloud-pubsub support this method to enable recovery mechanisms in applications?

Client#read_write_transaction is difficult to use.
I want to add a simpler API .

ex)

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    const DATABASE: &str = "projects/local-project/instances/test-instance/databases/local-database";
    let client = Client::new(DATABASE).await?;
    let retry = &mut TransactionRetry::new();
    loop {
        let tx = &mut client.begin_read_write_transaction().await?;
   
        let result = run_in_transaction(tx).await;
    
        // try to commit or rollback transaction.
        match tx.end(result, None).await {
             Ok((_commit_timestamp, success)) => return Ok(success),
             Err(err) => retry.next(err).await? // check retry
        }
    }
    Ok(())
}

I'm working on a Google pub/sub connector over at https://github.com/risingwavelabs/risingwave where sources can be added ad-hoc through a SQL interface. As such I feel it makes natural sense to not tie down credentials to files on the filesystem, but I don't think its possible yet with the google-cloud-auth crate. I can work on a PR to make this possible if it sounds good!

Thanks for providing this library! I've managed to set up a file down- and upload based on your latest changes from today.

Is there a way to set the Content Type other than application/octet-stream based on the filename/extension?

My publisher is successfully publishing messages, and my subscriber is receiving them. But every time a messaged is published, the publisher has an error: "failed to notify : id=". What causes this error?

Add example with real pubsub project config, this is the example in the documentation

use google_cloud_pubsub::client::{Client, ClientConfig};
use google_cloud_gax::cancel::CancellationToken;
use google_cloud_googleapis::pubsub::v1::PubsubMessage;
use google_cloud_pubsub::topic::TopicConfig;
use google_cloud_pubsub::subscription::SubscriptionConfig;
use google_cloud_gax::grpc::Status;
use tokio::task::JoinHandle;

// Client config
#[tokio::main]
async fn main() -> Result<(), Status> {

// Create pubsub client.
// `use google_cloud_default::WithAuthExt;` is required to use default authentication.
let config = ClientConfig::default();//.with_auth().await.unwrap();
let client = Client::new(config).await.unwrap();

// Create topic.
let topic = client.topic("test-topic");
if !topic.exists(None, None).await? {
    topic.create(None, None, None).await?;
}

// Start publisher.
let publisher = topic.new_publisher(None);

// Publish message.
let tasks : Vec<JoinHandle<Result<String,Status>>> = (0..10).into_iter().map(|_i| {
    let publisher = publisher.clone();
    tokio::spawn(async move {
        let mut msg = PubsubMessage::default();
        msg.data = "abc".into();
        // Set ordering_key if needed (https://cloud.google.com/pubsub/docs/ordering)
        // msg.ordering_key = "order".into();

        // Send a message. There are also `publish_bulk` and `publish_immediately` methods.
        let mut awaiter = publisher.publish(msg).await;

        // The get method blocks until a server-generated ID or an error is returned for the published message.
        awaiter.get(None).await
    })
}).collect();

// Wait for all publish task finish
for task in tasks {
    let message_id = task.await.unwrap()?;
}

// Wait for publishers in topic finish.
let mut publisher = publisher;
publisher.shutdown();

Ok(())

}

I have a config module that returns a Client:

mod config {
    use google_cloud_default::WithAuthExt;
    use google_cloud_storage::client::{Client, ClientConfig};

    pub async fn get_google_cloud_client() -> Client {
        let config = ClientConfig::default().with_auth().await.unwrap();
        Client::new(config)
    }
}

When I make API calls to Cloud Storage from my test module I get the following error: panicked at 'This is dummy token source provider. you can use 'google_cloud_default'. I initialised the client by calling the above getter.

 #[tokio::test]
 async fn test_call_to_gcs() {
    let client = config::get_google_cloud_client().await
    // Make API calls with this client...

    // ... then: panicked at 'This is dummy token source provider. you can use 'google_cloud_default
}

This problem occurred, when I was migrating to the latest version. However, I don't face this problem when I make API calls to Cloud Storage inside my main function, when running my program. In my main function I also created a client by initialising it to let client = config::get_google_cloud_client().await Currently I don't know how to fix this problem: How can I use the automatic authentication inside unit tests in test modules?

To repro:

• create a database and a change stream
• query the change stream in a tokio::select! loop like

loop {
    tokio::select! {
        _ = some_tokio_interval.tick() => {
            do_periodic_task();
        },
       next_entry_res = iterator.next() => {
         next_entry_res?;
        }
    }
}

• wait until it's getting heartbeats from the iterator
• delete the change stream

At this point, the iterator should error and we return from the loop. However the client goes into a retry, giving the tick() future a chance to complete. the iterator.next() is cancelled and the error never returned to next_entry_res. It goes into an error state,but this error state always returns None for next() calls. This is particularly bad for change streams, because None indicates you've read everything from the stream, which may not be the case. So this loops forever.

Rather than using a generic Error type would be helpful to implement a BQError enum similar to the below from https://github.com/lquerel/gcp-bigquery-client/blob/main/src/error.rs

pub enum BQError {
    #[error("Invalid service account key (error: {0})")]
    InvalidServiceAccountKey(#[from] std::io::Error),

    #[error("Invalid service account authenticator (error: {0})")]
    InvalidServiceAccountAuthenticator(std::io::Error),

    #[error("Invalid installed flow authenticator (error: {0})")]
    InvalidInstalledFlowAuthenticator(std::io::Error),

    #[error("Invalid installed application default credentials authenticator (error: {0})")]
    InvalidApplicationDefaultCredentialsAuthenticator(std::io::Error),

    #[error("Invalid authorized user authenticator (error: {0})")]
    InvalidAuthorizedUserAuthenticator(std::io::Error),

    #[error("Authentication error (error: {0})")]
    AuthError(#[from] yup_oauth2::error::AuthError),

    #[error("Authentication error (error: {0})")]
    YupAuthError(#[from] yup_oauth2::Error),

    #[error("No token")]
    NoToken,

    #[error("Request error (error: {0})")]
    RequestError(#[from] reqwest::Error),

    #[error("Response error (error: {error:?})")]
    ResponseError { error: ResponseError },

    #[error("No data available. The result set is positioned before the first or after the last row. Try to call the method next on your result set.")]
    NoDataAvailable,

    #[error("Invalid column index (col_index: {col_index})")]
    InvalidColumnIndex { col_index: usize },

    #[error("Invalid column name (col_name: {col_name})")]
    InvalidColumnName { col_name: String },

    #[error("Invalid column type (col_index: {col_index}, col_type: {col_type}, type_requested: {type_requested})")]
    InvalidColumnType {
        col_index: usize,
        col_type: String,
        type_requested: String,
    },

    #[error("Json serialization error (error: {0})")]
    SerializationError(#[from] serde_json::Error),
}

It seems that it's required to be using a service account for interacting with Cloud Storage. Is my understanding wrong?

Google's official SDKs allow authenticating without providing service account credentials so are you open to changing the behavior of (at least) google-cloud-storage to allow for creating a Client without using a service account?

Issue

When running the below code block (from the readme) to instantiate Client struct

let client = Client::default().await.unwrap();

I get the following error:

GAX(Auth(HttpError(reqwest::Error { kind: Decode, source: Error("EOF while parsing a value", line: 1, column: 0) })))

I am using the google_auth crate from this repository utilising the below code block to set the credentials (from the readme)

    let credentials_path = PathBuf::from("./src/keys/dev-key.json");
    let service_account = CustomServiceAccount::from_file(credentials_path).expect("Service account threw an error");
    let authentication_manager = AuthenticationManager::from(service_account);
    let scopes = &["https://www.googleapis.com/auth/cloud-platform"];

Setting the credentials seems to be working fine; when I run the code in debug, I can see the below all populating with the correct values

My environment

rustc --version
rustc 1.62.1 (e092d0b6b 2022-07-16)

cargo --version
cargo 1.62.1 (a748cf5a3 2022-06-08)

My Cargo.toml file looks like this:

actix-web = "4.1.0"
awc = "3.0.0"
colored = "2.0.0"
dotenv = "0.15.0"
env_logger = "0.9.0"
futures = "0.3.21"
gcp_auth = "0.7.3"
google-cloud-gax = "0.7.0"
google-cloud-googleapis = {version = "0.4.0", features = ["pubsub"]}
google-cloud-pubsub = {version = "0.4.0", features = ["trace"]}
google_auth = "0.2.1"
prost-build = "0.10.4"
protobuf = "3.1.0"
serde = { version = "1.0.140", features = ["derive"] }
serde_json = "1.0.82"
tokio = "1.20.1"
validator = "0.16.0"
validator_derive = "0.16.0"

Hi there,

I am trying to setup the CloudStorage client but I have the following error when trying out the library (copy-pasted the upload example).

Compilling with cargo run gives the following error :

    |
146 |     #[serde(default, with = "time::serde::rfc3339::option")]
    |                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ could not find `rfc3339` in `serde`

Am I missing something?

HI! After this merged, does lib supports gcp workload identity #105? Thanks

https://github.com/googleapis/google-cloud-go/blob/main/storage/bucket.go#L386

Because multiple requests might be required to complete a rewrite of a large object, you need to call the rewrite in a loop. The documentation could be improved to make this more obvious. Something like this:

use google_cloud_storage::client::Client;
use google_cloud_storage::http::objects::rewrite::RewriteObjectRequest;

#[tokio::main]
async fn main() {
    let mut done = false;
    let mut rewrite_token = None;
    let client = Client::default().await.unwrap();

    while !done {
        let result = client.rewrite_object(&RewriteObjectRequest{
            source_bucket: "bucket1".to_string(),
            source_object: "object".to_string(),
            destination_bucket: "bucket2".to_string(),
            destination_object: "object1".to_string(),
            rewrite_token: rewrite_token.clone(),
            ..Default::default()
        }, None).await;
        
        done = result.done;
        rewrite_token = result.rewrite_token;
    }
}

https://cloud.google.com/storage/docs/performing-resumable-uploads

Most of the client APIs appear to take an optional cancellation token. Tokio's CancellationToken type is designed to allow alerting a spawned task that cancellation is being requested. Because google-cloud-rust isn't spawning tasks, it's not necessary for a cancellation mechanism to be built in to the library. Instead, users of the library can simply drop the futures returned from the google-cloud-rust APIs, perhaps in response to a CancellationToken firing, to enable cancellation.

I want to be able to get all the objects from a particular bucket like the below documentation

https://cloud.google.com/storage/docs/json_api/v1/objects/list

Can this be implemented? I am trying to add it to the current code, but I need some help understanding how to go about this

Hi,

Thanks for this awesome work,
A quick question. Is there any way to query log-based metrics using this crate?

I get the following error:

cargo build
   Compiling pubsub-example v0.1.0 (/Users/aviramhassan/Code/mirrord-intellij/pubsub-example)
error[E0599]: no method named `with_auth` found for struct `ClientConfig` in the current scope
 --> src/main.rs:6:42
  |
6 |     let config = ClientConfig::default().with_auth().await.unwrap();
  |                                          ^^^^^^^^^ method not found in `ClientConfig`

warning: unused import: `google_cloud_default::WithAuthExt`
 --> src/main.rs:2:5
  |
2 | use google_cloud_default::WithAuthExt;
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |
  = note: `#[warn(unused_imports)]` on by default

For more information about this error, try `rustc --explain E0599`.
warning: `pubsub-example` (bin "pubsub-example") generated 1 warning
error: could not compile `pubsub-example` due to previous error; 1 warning emitted

src/main.rs:

use google_cloud_pubsub::client::{ClientConfig, Client};
use google_cloud_default::WithAuthExt;

#[tokio::main]
async fn main() {
    let config = ClientConfig::default().with_auth().await.unwrap();
    let client = Client::new(config).await.unwrap();
}

Cargo.toml:

[package]
name = "pubsub-example"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
google-cloud-pubsub = "0.15"
google-cloud-default = { version = "0.4", features = ["pubsub"] }
tokio = { version = "1.28.2", features = ["full"] }

cargo version:

cargo --version
cargo 1.69.0 (6e9a83356 2023-04-12)

thread 'main' panicked at 'called Result::unwrap() on an Err value: GAX(InvalidEmulatorHOST(""))', src/main.rs:15:45. How do i resolve this? I ran the code as is

I an trying to run the example from the repo:

use google_cloud_auth::credentials::CredentialsFile;
use google_cloud_storage::client::{ClientConfig, Client};
use google_cloud_default::WithAuthExt;

async fn run(cred: CredentialsFile) {
    let config = ClientConfig::default().with_credentials(cred).await.unwrap();
    let client = Client::new(config);
}

The Cargo.toml is:

[dependencies]
google-cloud-storage = "0.9.0"
google-cloud-auth = "0.9.0"
google-cloud-default = { version = "0.1.0", features = ["storage", "google-cloud-metadata"] }

I get the following error:

.with_credentials(cred))
^^^^^^^^^^^^^^^^ method not found in `ClientConfig`

can someone point what am I missing?

Seems like API allows to upload file with metadata by multipart.
reqwest provides some basic support for multiparts via multipart features which adds mime_guess dependency.

If I'm correct that's all there to it, would you mind having it?

Currently the Storage Client is not clone.
It also cannot be derived directly because ServiceAccountClient is also not clone
Can we derive that on these two structs or does that introduce sideeffects?
I would imagine that duplicating the ServiceAccountClient means duplicating the authentication logic.
Thoughts @yoshidan?

Currently pulling from a PubSub subscription with large messages (>4MB) will cause the stream to terminate. I have not tried but I believe publishing large messages (>4MB) to a topic will trigger the same stream termination.

The cause is a change in tonic 0.9.0, specifically this PR which adds a maximum encode/decode message size, which defaults to 4MB.

PubSub supports up to 10MB, so we should try update google-cloud-rust to specify a larger size when sending the stream request.

Basic Reproduction Case

Putting together something simple together and running with RUST_LOG=error cargo run

use futures_util::StreamExt;
use google_cloud_default::WithAuthExt;
use google_cloud_pubsub::client::{Client, ClientConfig};
use tracing_subscriber::{fmt, prelude::*, EnvFilter};

#[tokio::main]
async fn main() {
    tracing_subscriber::registry()
        .with(fmt::layer())
        .with(EnvFilter::from_default_env())
        .init();
        
    let config = ClientConfig::default().with_auth().await.unwrap();
    let client = Client::new(config).await.unwrap();
    
    let mut subscription = client.subscription("my-subscription");
    let mut messages = subscription.subscribe(None).await.unwrap();
    
    while let Some(msg) = messages.next().await {
        msg.ack().await.unwrap();
    }
}

Gives the following error:

ERROR google_cloud_pubsub::subscriber: terminated subscriber streaming with error Status { code: OutOfRange, message: "Error, message length too large: found 4750628 bytes, the limit is: 4194304 bytes", source: None } : projects/my-project/subscriptions/my-subscription

Which I've traced to this addition in the above PR.

If you use Client.Query(...) for a query which can take a while (>~10 seconds), BigQuery will respond with:

{
  "kind": "bigquery#queryResponse",
  "jobReference": {
    "projectId": "...",
    "jobId": "...",
    "location": "US"
  },
  "jobComplete": false
}

Currently it is expected this is a google_cloud_bigquery::http::job::query::QueryResponse but it fails to parse due to not containing every field expected and there is also no retry logic to recheck and await job completion.

The java sdk here add support for endpoint override.
This is a great feature to prevent internet traffic to pubsub from my machines when a google private service connect is configured.

Right now the connection manager is internal and the domain is hardcoded as 'static &str here

Is there a chance to add this feature?

Is there any way, currently, to associate metadata with an object? client::upload_object seems to explicitly not support attaching custom metadata to the object. Is it the case that either:

1. This is known and not planned to be supported (if so, why?)?
2. This is something that is planned to be supported but isn't, yet (and, so, a PR would be accepted)?
3. It already is supported, but with another set of methods and objects besides upload_object?

Thank you for open-sourcing this work!

Looking at the compose.rs, I see the build function doesn't give us the ability to specify a mime type. According th the google documentation, the mime type should be the type of the first object if not set, but this is not showing on the cloud console. Can you add the ability to specify a mime type?

pub(crate) fn build(base_url: &str, client: &Client, req: &ComposeObjectRequest) -> RequestBuilder {
    let url = format!(
        "{}/b/{}/o/{}/compose",
        base_url,
        req.bucket.escape(),
        req.destination_object.escape()
    );
    let builder = client.post(url).query(&req).json(&req.composing_targets);
    if let Some(e) = &req.encryption {
        e.with_headers(builder)
    } else {
        builder
    }
}

Error: HttpClient(reqwest::Error { kind: Decode, source: Error("missing field crc32c", line: 23, column: 5) })

    let list_result = client
        .list_objects(&ListObjectsRequest {
            bucket: "bucketname".to_string(),
            ..Default::default()
        })
        .await?;

The code above fails to properly decode the response. It's requiring fields that are available only when getting a specific object.

Typical JSON Response:

{
  "items": [
    {
      "eventBasedHold": false, 
      "kind": "storage#object", 
      "contentType": "text/plain", 
      "name": "thumbnails/", 
      "etag": "CP+Qir3Onf4CEAE=", 
      "generation": "1681071265712255", 
      "temporaryHold": false, 
      "bucket": "bucketname", 
      "updated": "2023-04-09T20:14:25.713Z", 
      "metageneration": "1", 
      "mediaLink": "https://storage.googleapis.com/download/storage/v1/b/bucketname/o/thumbnails%2F?generation=1681071265712255&alt=media", 
      "kmsKeyName": "kmskey/path/name", 
      "timeStorageClassUpdated": "2023-04-09T20:14:25.713Z", 
      "size": "0", 
      "timeCreated": "2023-04-09T20:14:25.713Z", 
      "id": "bucketname/thumbnails//1681071265712255", 
      "selfLink": "https://www.googleapis.com/storage/v1/b/bucketname/o/thumbnails%2F", 
      "storageClass": "STANDARD"
    }, 
    {
      "kind": "storage#object", 
      "contentType": "image/png", 
      "name": "thumbnails/Screenshot 2023-01-27 at 11.54.48 AM.png", 
      "etag": "CJzcx/nSnf4CEAE=", 
      "generation": "1681072466292252", 
      "bucket": "bucketname", 
      "updated": "2023-04-09T20:34:26.293Z", 
      "metageneration": "1", 
      "mediaLink": "https://storage.googleapis.com/download/storage/v1/b/bucketname/o/thumbnails%2FScreenshot%202023-01-27%20at%2011.54.48%20AM.png?generation=1681072466292252&alt=media", 
      "kmsKeyName": "kmskey/path/name", 
      "timeStorageClassUpdated": "2023-04-09T20:34:26.293Z", 
      "size": "207122", 
      "timeCreated": "2023-04-09T20:34:26.293Z", 
      "id": "bucketname/thumbnails/Screenshot 2023-01-27 at 11.54.48 AM.png/1681072466292252", 
      "selfLink": "https://www.googleapis.com/storage/v1/b/bucketname/o/thumbnails%2FScreenshot%202023-01-27%20at%2011.54.48%20AM.png", 
      "storageClass": "STANDARD"
    }
  ], 
  "kind": "storage#objects"
}

Need to define a struct to handle for list object's case or make some of the existing fields in Object optional.

Hello!

First of all, thanks for working on this library! I was wondering if you're thinking of adding Datastore support in the future releases. I'm currently using google-cloud-rs however it is not really maintained actively and missing some features (e.g. unindexed values).

It would be really nice to have a better alternative to google-cloud-rs and I guess you are on it 🐻

GCS object size support is not tied to the word size of the client, so types like ChunkSize and TotalSize should use u64 instead of usize.

The gRPC api seems supported in Golang implementation at version 1.23 experimentally.

https://github.com/googleapis/google-cloud-go/blob/main/storage/storage.go#L131

I am trying to figure how to upgrade from 0.13.0 to 0.14.0 and having issues with the client config of the pubsub crate.

The method .with_auth() cannot be resolved anymore after the upgrade, and i cannot find any example on how to deal with the authentication with the newer version.

I have seen tickets making the authentication part optional and others to provide your own implementation, but why is this one gone ?
Could you please:

• explain the changes
• document it
• update the given example with updated code

Using:

[dependencies]
google-cloud-storage = "0.6.1"
tokio = { version = "1.23.0", features=["rt-multi-thread"]}

We get:

error[E0432]: unresolved import `google_cloud_storage::client::http`

It seems maybe the hardcoded dependency on default SA email prevents us from using this in our production environment using Workload Identity for application SA.

Do you accept PRs?

I'm working on a high-perf requirement pubsub client where I've been told it would be unacceptable to individually ack messages. I wasn't able to see examples for batch acking here, and from what I see going on internally in ReceivedMessage.ack() (reference), it calls the crate-private SubscriberClient.acknowledge() method which supports batches. Is this API exposed publicly somewhere in the package or ack-batching taken care of in some other way?

I've been trying to use user account token for authentication while developing locally, but it fails with an error.

I have a valid token in ~/.config/gcloud/application_default_credentials.json, created by running gcloud auth application-default login. When running the sample code

let client = google_cloud_pubsub::client::Client::default().await?;

it triggers the following error:

Error: http error

Caused by:
    0: error decoding response body: EOF while parsing a value at line 1 column 0
    1: EOF while parsing a value at line 1 column 0

Digging a little under the hood, the problem seems to be the response of UserAccountTokenSource requesting PUT https://oauth2.googleapis.com/token, which returns a 404 and an empty response body. Changing this to POST seems to make it work.

I would like to use a syntax like client.query(project_id, query).use_storage_read(true) or client.query_with_storage(project_id, query) to send a query and then read the results via the storage read api.

Currently we need to use client.job.query(...), wait for the job to finish, grab the destination id (if it is temporary/unknown) from the response via client.job().get(...) and then initiate a client.read_table(...) to pull the data. This is fine but I imagine that this is a common enough use case to warrant providing a convenient top level api.

If you could outline the which api interface you think is best I'd be happy to have a go making a PR.

Hi,
I am trying the library to store large file > 2GB and have trouble doing the same, here is my code for reference:

use google_cloud_default::WithAuthExt;
use google_cloud_storage::client::{Client, ClientConfig};
use google_cloud_storage::http::objects::upload::{Media, UploadObjectRequest, UploadType};

#[tokio::main]
async fn main() {
    let config = ClientConfig::default().with_auth().await
        .expect("failed getting credential");
    let client = Client::new(config);
    let data = vec![1; 3_000_000_000];
    println!("Storing {:.2}GB of file", data.len() as f64 / 1073741824f64);
    let upload_type = UploadType::Simple(Media::new("test_file.bin"));
    client
        .upload_object(
            &UploadObjectRequest {
                bucket: "my_bucket_name".to_string(),
                ..Default::default()
            },
            data,
            &upload_type,
            None
        )
        .await.expect("failed storing file");
    println!("Stored successfully");
}

this doesn't work the upload seems to be hung, not network activity have waited for over 2hrs.
whereas if I reduce use this data to less than 2GB let data = vec![1; 2_000_000_000]; it works?

Any ideas to troubleshoot this would be really helpful?

Description

Right now, Subscription.subscribe create an async_channel::unbounded::<ReceivedMessage>(); which is working well, however, I think we should have an option to set the size of the channel.

I think there is high chance of OOM, if our google pubsub consumer is slower than the publisher. or if the publisher sometimes, submit a huge amount of data.

Firstly, thank you so much for this library; it's great and very straightforward to use (I've used the pub-sub crate).

I think BigQuery is one of the most used products inside GCP, so it would be super helpful to have a native rust crate to interact with BQ. Thanks again 👍

I'm wondering if you would be open to making the auth crate optional, so that is possible to provide your own auth implementation?

We use our own auth crate for all of our other gcp-related auth. It implements a sans-io approach so that we can control what http client is being used so that we can have the same http client throughout all of our projects.

We are trying to keep our dependency tree down as much as possible (as you might have noticed from my previous contributions), and for us it would be great to be able to use the same gcp auth implementation for all of our dependencies as well.

If you are open to it, I'm of course happy to do the work and implement it. But I wanted to check first that you are willing to merge such a feature.

I haven't looked into how to implement it, but I'm thinking that I would put the auth crate/implementation behind a feature (that is part of the default features), and then use some a trait or something that can be implement by other auth crates.

But I'll properly look into how to implement it if you think this is something you would accept.

When using the following dependencies:

google-cloud-storage = {version = "0.10.0", default-features = false, features = ["rustls-tls"]}
google-cloud-default = {version = "0.1.1", default-features = false, features = ["storage", "rustls-tls"]}

the compiled binary still links against libssl.so.

I'm definitely not an expert in Rust, but it looks like this is caused by the lack of default-features = false here. I don't know if any other changes will be necessary.