feat: Add CredentialsProvider to Publisher and Subscriber settings #475
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,21 +16,32 @@ | |
|
|
||
| package com.google.cloud.pubsublite.cloudpubsub; | ||
|
|
||
| import static com.google.cloud.pubsublite.internal.ExtractStatus.toCanonical; | ||
| import static com.google.cloud.pubsublite.internal.wire.ServiceClients.addDefaultMetadata; | ||
| import static com.google.cloud.pubsublite.internal.wire.ServiceClients.addDefaultSettings; | ||
|
|
||
| import com.google.api.gax.batching.BatchingSettings; | ||
| import com.google.api.gax.core.CredentialsProvider; | ||
| import com.google.api.gax.rpc.ApiException; | ||
| import com.google.auto.value.AutoValue; | ||
| import com.google.cloud.pubsublite.AdminClient; | ||
| import com.google.cloud.pubsublite.AdminClientSettings; | ||
| import com.google.cloud.pubsublite.Constants; | ||
| import com.google.cloud.pubsublite.Message; | ||
| import com.google.cloud.pubsublite.MessageTransformer; | ||
| import com.google.cloud.pubsublite.Partition; | ||
| import com.google.cloud.pubsublite.TopicPath; | ||
| import com.google.cloud.pubsublite.cloudpubsub.internal.WrappingPublisher; | ||
| import com.google.cloud.pubsublite.internal.wire.PartitionCountWatchingPublisherSettings; | ||
| import com.google.cloud.pubsublite.internal.wire.PubsubContext; | ||
| import com.google.cloud.pubsublite.internal.wire.PubsubContext.Framework; | ||
| import com.google.cloud.pubsublite.internal.wire.RoutingMetadata; | ||
| import com.google.cloud.pubsublite.internal.wire.SinglePartitionPublisherBuilder; | ||
| import com.google.cloud.pubsublite.v1.AdminServiceClient; | ||
| import com.google.cloud.pubsublite.v1.AdminServiceSettings; | ||
| import com.google.cloud.pubsublite.v1.PublisherServiceClient; | ||
| import com.google.cloud.pubsublite.v1.PublisherServiceSettings; | ||
| import com.google.common.annotations.VisibleForTesting; | ||
| import com.google.pubsub.v1.PubsubMessage; | ||
| import java.util.Optional; | ||
| import java.util.function.Supplier; | ||
|
|
@@ -65,17 +76,26 @@ public abstract class PublisherSettings { | |
| /** Batching settings for this publisher to use. Apply per-partition. */ | ||
| abstract Optional<BatchingSettings> batchingSettings(); | ||
|
|
||
| /** A provider for credentials. */ | ||
| abstract CredentialsProvider credentialsProvider(); | ||
|
|
||
| /** | ||
| * A supplier for new PublisherServiceClients. Should return a new client each time. If present, | ||
| * ignores CredentialsProvider. | ||
| */ | ||
| abstract Optional<Supplier<PublisherServiceClient>> serviceClientSupplier(); | ||
|
|
||
| /** The AdminClient to use, if provided. */ | ||
| abstract Optional<AdminClient> adminClient(); | ||
|
|
||
| // For testing. | ||
| abstract SinglePartitionPublisherBuilder.Builder underlyingBuilder(); | ||
|
|
||
| /** Get a new builder for a PublisherSettings. */ | ||
| public static Builder newBuilder() { | ||
| return new AutoValue_PublisherSettings.Builder() | ||
| .setCredentialsProvider( | ||
| PublisherServiceSettings.defaultCredentialsProviderBuilder().build()) | ||
| .setUnderlyingBuilder(SinglePartitionPublisherBuilder.newBuilder()); | ||
| } | ||
|
|
||
|
|
@@ -97,18 +117,61 @@ public abstract Builder setMessageTransformer( | |
| /** Batching settings for this publisher to use. Apply per-partition. */ | ||
| public abstract Builder setBatchingSettings(BatchingSettings batchingSettings); | ||
|
|
||
| /** A provider for credentials. */ | ||
| public abstract Builder setCredentialsProvider(CredentialsProvider credentialsProvider); | ||
|
|
||
| /** | ||
| * A supplier for new PublisherServiceClients. Should return a new client each time. If present, | ||
| * ignores CredentialsProvider. | ||
| */ | ||
| public abstract Builder setServiceClientSupplier(Supplier<PublisherServiceClient> supplier); | ||
|
|
||
| /** The AdminClient to use, if provided. */ | ||
| public abstract Builder setAdminClient(AdminClient adminClient); | ||
|
|
||
| // For testing. | ||
| @VisibleForTesting | ||
| abstract Builder setUnderlyingBuilder( | ||
| SinglePartitionPublisherBuilder.Builder underlyingBuilder); | ||
|
|
||
| public abstract PublisherSettings build(); | ||
| } | ||
|
|
||
| private PublisherServiceClient newServiceClient(Partition partition) throws ApiException { | ||
| if (serviceClientSupplier().isPresent()) return serviceClientSupplier().get().get(); | ||
| PublisherServiceSettings.Builder settingsBuilder = PublisherServiceSettings.newBuilder(); | ||
| settingsBuilder = settingsBuilder.setCredentialsProvider(credentialsProvider()); | ||
| settingsBuilder = | ||
| addDefaultMetadata( | ||
| PubsubContext.of(FRAMEWORK), | ||
| RoutingMetadata.of(topicPath(), partition), | ||
| settingsBuilder); | ||
| try { | ||
| return PublisherServiceClient.create( | ||
| addDefaultSettings(topicPath().location().region(), settingsBuilder)); | ||
| } catch (Throwable t) { | ||
| throw toCanonical(t).underlying; | ||
| } | ||
| } | ||
|
|
||
| private AdminClient getAdminClient() throws ApiException { | ||
| if (adminClient().isPresent()) return adminClient().get(); | ||
| try { | ||
| return AdminClient.create( | ||
| AdminClientSettings.newBuilder() | ||
| .setServiceClient( | ||
| AdminServiceClient.create( | ||
| addDefaultSettings( | ||
| topicPath().location().region(), | ||
| AdminServiceSettings.newBuilder() | ||
| .setCredentialsProvider(credentialsProvider())))) | ||
| .setRegion(topicPath().location().region()) | ||
| .build()); | ||
| } catch (Throwable t) { | ||
| throw toCanonical(t).underlying; | ||
| } | ||
| } | ||
|
|
||
| @SuppressWarnings("CheckReturnValue") | ||
| Publisher instantiate() throws ApiException { | ||
| BatchingSettings batchingSettings = batchingSettings().orElse(DEFAULT_BATCHING_SETTINGS); | ||
|
|
@@ -125,16 +188,12 @@ Publisher instantiate() throws ApiException { | |
| SinglePartitionPublisherBuilder.Builder singlePartitionBuilder = | ||
| underlyingBuilder() | ||
| .setBatchingSettings(batchingSettings) | ||
|
Member
There was a problem hiding this comment. Beam also touches this removed setContext method. Should it just remove the usage?
Member
There was a problem hiding this comment. This has been moved to newServiceClient method. (Thanks @dpcollins-google !) |
||
| .setTopic(topicPath()) | ||
| .setPartition(partition) | ||
| .setServiceClient(newServiceClient(partition)); | ||
| return singlePartitionBuilder.build(); | ||
| }) | ||
| .setAdminClient(getAdminClient()); | ||
| return new WrappingPublisher(publisherSettings.build().instantiate(), messageTransformer); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you instead add a specific exclusion for what you are adding? Not all changes to Builders will be adding abstract methods. Will this trigger on removing methods?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This specifically triggers on adding methods to nested classes named "Builder". I think thats specific enough. (code 7013 is added abstract method to abstract class)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No, this isn't specific enough and would require a major version bump if this API had reached 1.0.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I disagree. Addition of a method to a Builder class, which is only overridden in an autogenerated AutoValue file, would not trigger a major version bump as it is not a backwards-incompatible API surface change.