diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateTopicAndPublishMessages.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateTopicAndPublishMessages.java index fa243497c638..e39fa4d4e8e3 100644 --- a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateTopicAndPublishMessages.java +++ b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateTopicAndPublishMessages.java @@ -17,6 +17,7 @@ package com.google.cloud.examples.pubsub.snippets; import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; import com.google.cloud.pubsub.spi.v1.Publisher; import com.google.cloud.pubsub.spi.v1.TopicAdminClient; import com.google.protobuf.ByteString; @@ -31,30 +32,54 @@ * publish messages to it. */ public class CreateTopicAndPublishMessages { - public static void main(String... args) throws Exception { + + public static void createTopic() throws Exception { TopicName topic = TopicName.create("test-project", "test-topic"); try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) { topicAdminClient.createTopic(topic); } + } + public static void publishMessages() throws Exception { + // [START publish] + TopicName topicName = TopicName.create("test-project", "test-topic"); Publisher publisher = null; + List> messageIdFutures = new ArrayList<>(); + try { - publisher = Publisher.defaultBuilder(topic).build(); + // Create a publisher instance with default settings bound to the topic + publisher = Publisher.defaultBuilder(topicName).build(); + List messages = Arrays.asList("first message", "second message"); - List> messageIds = new ArrayList<>(); + + // schedule publishing one message at a time : messages get automatically batched for (String message : messages) { ByteString data = ByteString.copyFromUtf8(message); + // message data is converted to base64-encoding PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build(); + + // Once published, returns a server-assigned message id (unique within the topic) ApiFuture messageIdFuture = publisher.publish(pubsubMessage); - messageIds.add(messageIdFuture); - } - for (ApiFuture messageId : messageIds) { - System.out.println("published with message ID: " + messageId.get()); + messageIdFutures.add(messageIdFuture); } } finally { + // wait on any pending publish requests. + List messageIds = ApiFutures.allAsList(messageIdFutures).get(); + + for (String messageId : messageIds) { + System.out.println("published with message ID: " + messageId); + } + if (publisher != null) { + // When finished with the publisher, shutdown to free up resources. publisher.shutdown(); } } + // [END publish] + } + + public static void main(String... args) throws Exception { + createTopic(); + publishMessages(); } } diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/PublisherSnippets.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/PublisherSnippets.java index 220b1c0b0b9d..f3f8c85042f8 100644 --- a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/PublisherSnippets.java +++ b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/PublisherSnippets.java @@ -24,10 +24,24 @@ import com.google.api.core.ApiFuture; import com.google.api.core.ApiFutureCallback; import com.google.api.core.ApiFutures; +import com.google.api.gax.batching.BatchingSettings; +import com.google.api.gax.batching.FlowControlSettings; +import com.google.api.gax.batching.FlowController.LimitExceededBehavior; +import com.google.api.gax.core.CredentialsProvider; +import com.google.api.gax.core.FixedCredentialsProvider; +import com.google.api.gax.grpc.ChannelProvider; +import com.google.api.gax.grpc.ExecutorProvider; +import com.google.api.gax.grpc.InstantiatingExecutorProvider; +import com.google.api.gax.retrying.RetrySettings; +import com.google.auth.oauth2.ServiceAccountCredentials; import com.google.cloud.pubsub.spi.v1.Publisher; +import com.google.cloud.pubsub.spi.v1.TopicAdminSettings; import com.google.protobuf.ByteString; import com.google.pubsub.v1.PubsubMessage; import com.google.pubsub.v1.TopicName; +import org.threeten.bp.Duration; + +import java.io.FileInputStream; /** This class contains snippets for the {@link Publisher} interface. */ public class PublisherSnippets { @@ -41,7 +55,6 @@ public PublisherSnippets(Publisher publisher) { // [TARGET publish(PubsubMessage)] // [VARIABLE "my_message"] public ApiFuture publish(String message) { - // [START publish] ByteString data = ByteString.copyFromUtf8(message); PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build(); ApiFuture messageIdFuture = publisher.publish(pubsubMessage); @@ -54,7 +67,6 @@ public void onFailure(Throwable t) { System.out.println("failed to publish: " + t); } }); - // [END publish] return messageIdFuture; } @@ -62,9 +74,8 @@ public void onFailure(Throwable t) { // [TARGET newBuilder(TopicName)] // [VARIABLE "my_project"] // [VARIABLE "my_topic"] - public static void newBuilder(String projectName, String topicName) throws Exception { - // [START newBuilder] - TopicName topic = TopicName.create(projectName, topicName); + public static void newBuilder(String projectId, String topicId) throws Exception { + TopicName topic = TopicName.create(projectId, topicId); Publisher publisher = Publisher.defaultBuilder(topic).build(); try { // ... @@ -72,6 +83,96 @@ public static void newBuilder(String projectName, String topicName) throws Excep // When finished with the publisher, make sure to shutdown to free up resources. publisher.shutdown(); } - // [END newBuilder] + } + + public Publisher getPublisherWithCustomBatchSettings(TopicName topicName) throws Exception { + // [START publisherBatchSettings] + // Batch settings control how the publisher batches messages + long requestBytesThreshold = 5000L; // default : 1kb + long messageCountBatchSize = 10L; // default : 100 + + Duration publishDelayThreshold = Duration.ofMillis(100); // default : 1 ms + + // Publish request get triggered based on request size, messages count & time since last publish + BatchingSettings batchingSettings = BatchingSettings.newBuilder() + .setElementCountThreshold(messageCountBatchSize) + .setRequestByteThreshold(requestBytesThreshold) + .setDelayThreshold(publishDelayThreshold) + .build(); + + Publisher publisher = Publisher.defaultBuilder(topicName) + .setBatchingSettings(batchingSettings).build(); + // [END publisherBatchSettings] + return publisher; + } + + public Publisher getPublisherWithCustomRetrySettings(TopicName topicName) throws Exception { + // [START publisherRetrySettings] + // Retry settings control how the publisher handles retryable failures + Duration retryDelay = Duration.ofMillis(100); // default : 1 ms + double retryDelayMultiplier = 2.0; // back off for repeated failures + Duration maxRetryDelay = Duration.ofSeconds(5); // default : 10 seconds + + RetrySettings retrySettings = RetrySettings.newBuilder() + .setInitialRetryDelay(retryDelay) + .setRetryDelayMultiplier(retryDelayMultiplier) + .setMaxRetryDelay(maxRetryDelay) + .build(); + + Publisher publisher = Publisher.defaultBuilder(topicName) + .setRetrySettings(retrySettings).build(); + // [END publisherRetrySettings] + return publisher; + } + + public Publisher getPublisherWithCustomFlowControlSettings(TopicName topicName) throws Exception { + // [START publisherFlowControlSettings] + + // Flow control settings restrict the number of outstanding publish requests + int maxOutstandingBatches = 20; + int maxOutstandingRequestBytes = 500000; + + // override behavior on limits exceeded if needed, default behavior is to block + LimitExceededBehavior limitExceededBehavior = LimitExceededBehavior.ThrowException; + + FlowControlSettings flowControlSettings = FlowControlSettings.newBuilder() + .setMaxOutstandingElementCount(maxOutstandingBatches) + .setMaxOutstandingRequestBytes(maxOutstandingRequestBytes) + .setLimitExceededBehavior(limitExceededBehavior) + .build(); + + Publisher publisher = Publisher.defaultBuilder(topicName) + .setFlowControlSettings(flowControlSettings).build(); + // [END publisherFlowControlSettings] + return publisher; + } + + public Publisher getSingleThreadedPublisher(TopicName topicName) throws Exception { + // [START singleThreadedPublisher] + // create a publisher with a single threaded executor + ExecutorProvider executorProvider = InstantiatingExecutorProvider.newBuilder() + .setExecutorThreadCount(1).build(); + Publisher publisher = Publisher.defaultBuilder(topicName) + .setExecutorProvider(executorProvider).build(); + // [END singleThreadedPublisher] + return publisher; + } + + private Publisher createPublisherWithCustomCredentials(TopicName topicName) throws Exception { + // [START publisherWithCustomCredentials] + // read service account credentials from file + CredentialsProvider credentialsProvider = + FixedCredentialsProvider + .create(ServiceAccountCredentials.fromStream( + new FileInputStream("credentials.json"))); + ChannelProvider channelProvider = + TopicAdminSettings.defaultChannelProviderBuilder() + .setCredentialsProvider(credentialsProvider).build(); + + Publisher publisher = Publisher.defaultBuilder(topicName) + .setChannelProvider(channelProvider) + .build(); + // [START publisherWithCustomCredentials] + return publisher; } } diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriberSnippets.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriberSnippets.java index bfd495b0cf78..2c5756078b02 100644 --- a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriberSnippets.java +++ b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriberSnippets.java @@ -23,39 +23,46 @@ package com.google.cloud.examples.pubsub.snippets; import com.google.api.core.ApiFuture; +import com.google.api.gax.batching.FlowControlSettings; +import com.google.api.gax.core.CredentialsProvider; +import com.google.api.gax.core.FixedCredentialsProvider; +import com.google.api.gax.grpc.ChannelProvider; +import com.google.api.gax.grpc.ExecutorProvider; +import com.google.api.gax.grpc.InstantiatingExecutorProvider; +import com.google.auth.oauth2.ServiceAccountCredentials; import com.google.cloud.pubsub.spi.v1.AckReplyConsumer; import com.google.cloud.pubsub.spi.v1.MessageReceiver; import com.google.cloud.pubsub.spi.v1.Subscriber; +import com.google.cloud.pubsub.spi.v1.TopicAdminSettings; +import com.google.common.util.concurrent.MoreExecutors; import com.google.pubsub.v1.PubsubMessage; import com.google.pubsub.v1.SubscriptionName; +import java.io.FileInputStream; import java.util.concurrent.Executor; /** This class contains snippets for the {@link Subscriber} interface. */ public class SubscriberSnippets { - private final SubscriptionName subscription; + private final SubscriptionName subscriptionName; private final MessageReceiver receiver; private final ApiFuture done; private final Executor executor; public SubscriberSnippets( - SubscriptionName subscription, + SubscriptionName subscriptionName, MessageReceiver receiver, ApiFuture done, Executor executor) { - this.subscription = subscription; + this.subscriptionName = subscriptionName; this.receiver = receiver; this.done = done; this.executor = executor; } - /** - * Example of receiving a specific number of messages. - */ // [TARGET startAsync()] public void startAndWait() throws Exception { // [START startAsync] - Subscriber subscriber = Subscriber.defaultBuilder(subscription, receiver).build(); + Subscriber subscriber = Subscriber.defaultBuilder(subscriptionName, receiver).build(); subscriber.addListener(new Subscriber.Listener() { public void failed(Subscriber.State from, Throwable failure) { // Handle error. @@ -70,14 +77,13 @@ public void failed(Subscriber.State from, Throwable failure) { } private void createSubscriber() throws Exception { - // [START async_pull_subscription] + // [START pullSubscriber] String projectId = "my-project-id"; String subscriptionId = "my-subscription-id"; - SubscriptionName subscriptionName = SubscriptionName.create(projectId, subscriptionId); + SubscriptionName subscriptionName = SubscriptionName.create(projectId, subscriptionId); // Instantiate an asynchronous message receiver - MessageReceiver receiver = - new MessageReceiver() { + MessageReceiver receiver = new MessageReceiver() { @Override public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) { // handle incoming message, then ack or nack the received message @@ -98,7 +104,63 @@ public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) { subscriber.stopAsync(); } } - // [END async_pull_subscription] + // [END pullSubscriber] + } + + private Subscriber createSubscriberWithErrorListener() throws Exception { + // [START subscriberWithErrorListener] + Subscriber subscriber = Subscriber.defaultBuilder(subscriptionName, receiver).build(); + + subscriber.addListener(new Subscriber.Listener() { + public void failed(Subscriber.State from, Throwable failure) { + // Handle error. + } + }, MoreExecutors.directExecutor()); + // [END subscriberWithErrorListener] + return subscriber; + } + + private Subscriber createSingleThreadedSubscriber() throws Exception { + // [START singleThreadedSubscriber] + // provide a separate executor service for polling + ExecutorProvider executorProvider = InstantiatingExecutorProvider.newBuilder() + .setExecutorThreadCount(1).build(); + + Subscriber subscriber = Subscriber.defaultBuilder(subscriptionName, receiver) + .setExecutorProvider(executorProvider) + .build(); + // [END singleThreadedSubscriber] + return subscriber; } + private Subscriber createSubscriberWithCustomFlowSettings() throws Exception { + // [START subscriberWithCustomFlow] + int maxMessageCount = 10; + // Configure max number of messages to be pulled + FlowControlSettings flowControlSettings = FlowControlSettings.newBuilder() + .setMaxOutstandingElementCount(maxMessageCount) + .build(); + Subscriber subscriber = Subscriber.defaultBuilder(subscriptionName, receiver) + .setFlowControlSettings(flowControlSettings) + .build(); + // [END subscriberWithCustomFlow] + return subscriber; + } + + private Subscriber createSubscriberWithCustomCredentials() throws Exception { + // [START subscriberWithCustomCredentials] + CredentialsProvider credentialsProvider = + FixedCredentialsProvider + .create(ServiceAccountCredentials.fromStream( + new FileInputStream("credentials.json"))); + ChannelProvider channelProvider = + TopicAdminSettings.defaultChannelProviderBuilder() + .setCredentialsProvider(credentialsProvider).build(); + + Subscriber subscriber = Subscriber.defaultBuilder(subscriptionName, receiver) + .setChannelProvider(channelProvider) + .build(); + // [START subscriberWithCustomCredentials] + return subscriber; + } } diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionAdminClientSnippets.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionAdminClientSnippets.java index 2be344bea197..015c75316b7f 100644 --- a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionAdminClientSnippets.java +++ b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionAdminClientSnippets.java @@ -54,7 +54,7 @@ public String getProjectId() { /** Example of creating a pull subscription for a topic. */ public Subscription createSubscription(String topicId, String subscriptionId) throws Exception { - // [START createSubscription] + // [START createPullSubscription] try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) { // eg. projectId = "my-test-project", topicId = "my-test-topic" TopicName topicName = TopicName.create(projectId, topicId); @@ -67,13 +67,14 @@ public Subscription createSubscription(String topicId, String subscriptionId) th subscriptionName, topicName, PushConfig.getDefaultInstance(), 0); return subscription; } - // [END createSubscription] + // [END createPullSubscription] } /** Example of creating a subscription with a push endpoint. */ - public Subscription createSubscriptionWithPushEndpoint(String topicId, String subscriptionId, String endpoint) + public Subscription createSubscriptionWithPushEndpoint(String topicId, String subscriptionId, + String endpoint) throws Exception { - // [START createSubscriptionWithPushEndpoint] + // [START createPushSubscription] try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) { TopicName topicName = TopicName.create(projectId, topicId); SubscriptionName subscriptionName = @@ -90,7 +91,7 @@ public Subscription createSubscriptionWithPushEndpoint(String topicId, String su subscriptionName, topicName, pushConfig, ackDeadlineInSeconds); return subscription; } - // [END createSubscriptionWithPushEndpoint] + // [END createPushSubscription] } /** Example of replacing the push configuration of a subscription, setting the push endpoint. */ diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/TopicAdminClientSnippets.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/TopicAdminClientSnippets.java index 4a4394491e71..dd9af0eed299 100644 --- a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/TopicAdminClientSnippets.java +++ b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/TopicAdminClientSnippets.java @@ -50,6 +50,8 @@ public String getProjectId() { public Topic createTopic(String topicId) throws Exception { // [START createTopic] try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) { + // projectId <= unique project identifier, eg. "my-project-id" + // topicId <= "my-topic-id" TopicName topicName = TopicName.create(projectId, topicId); Topic topic = topicAdminClient.createTopic(topicName); return topic;