add back PubSub samples (#1540)

pongad · web-flow · commit d70aaf7f9818 · 2017-01-19T12:00:59.000+11:00
* add back PubSub samples
diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2017 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.examples.pubsub.snippets;
+
+import com.google.cloud.pubsub.spi.v1.MessageReceiver;
+import com.google.cloud.pubsub.spi.v1.Subscriber;
+import com.google.cloud.pubsub.spi.v1.SubscriberClient;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Service;
+import com.google.pubsub.v1.PubsubMessage;
+import com.google.pubsub.v1.PushConfig;
+import com.google.pubsub.v1.SubscriptionName;
+import com.google.pubsub.v1.TopicName;
+
+/**
+ * A snippet for Google Cloud Pub/Sub showing how to create a Pub/Sub pull subscription and
+ * asynchronously pull messages from it.
+ */
+public class CreateSubscriptionAndPullMessages {
+
+  public static void main(String... args) throws Exception {
+    TopicName topic = TopicName.create("test-project", "test-topic");
+    SubscriptionName subscription = SubscriptionName.create("test-project", "test-subscription");
+
+    try (SubscriberClient subscriberClient = SubscriberClient.create()) {
+      subscriberClient.createSubscription(subscription, topic, PushConfig.getDefaultInstance(), 0);
+    }
+
+    MessageReceiver receiver =
+        new MessageReceiver() {
+          @Override
+          public ListenableFuture<MessageReceiver.AckReply> receiveMessage(PubsubMessage message) {
+            System.out.println("got message: " + message.getData().toStringUtf8());
+            return Futures.immediateFuture(MessageReceiver.AckReply.ACK);
+          }
+        };
+    Subscriber subscriber = null;
+    try {
+      subscriber = Subscriber.Builder.newBuilder(subscription, receiver).build();
+      subscriber.addListener(
+          new Service.Listener() {
+            @Override
+            public void failed(Service.State from, Throwable failure) {
+              System.err.println(failure);
+            }
+          },
+          MoreExecutors.directExecutor());
+      subscriber.startAsync().awaitRunning();
+      Thread.sleep(60000);
+    } finally {
+      if (subscriber != null) {
+        subscriber.stopAsync();
+      }
+    }
+  }
+}
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
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2017 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.examples.pubsub.snippets;
+
+import com.google.cloud.pubsub.spi.v1.Publisher;
+import com.google.cloud.pubsub.spi.v1.PublisherClient;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
+import com.google.pubsub.v1.PubsubMessage;
+import com.google.pubsub.v1.TopicName;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A snippet for Google Cloud Pub/Sub showing how to create a Pub/Sub topic and asynchronously
+ * publish messages to it.
+ */
+public class CreateTopicAndPublishMessages {
+  public static void main(String... args) throws Exception {
+    TopicName topic = TopicName.create("test-project", "test-topic");
+    try (PublisherClient publisherClient = PublisherClient.create()) {
+      publisherClient.createTopic(topic);
+    }
+
+    Publisher publisher = null;
+    try {
+      publisher = Publisher.Builder.newBuilder(topic).build();
+      List<String> messages = Arrays.asList("first message", "second message");
+      List<ListenableFuture<String>> messageIds = new ArrayList<>();
+      for (String message : messages) {
+        ByteString data = ByteString.copyFromUtf8(message);
+        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();
+        ListenableFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
+        messageIds.add(messageIdFuture);
+      }
+      for (String messageId : Futures.allAsList(messageIds).get()) {
+        System.out.println("published with message ID: " + messageId);
+      }
+    } finally {
+      if (publisher != null) {
+        publisher.shutdown();
+      }
+    }
+  }
+}
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageReceiver.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageReceiver.java
@@ -20,7 +20,7 @@
 import com.google.pubsub.v1.PubsubMessage;
 
 /** Users of the {@link Subscriber} must implement this interface to receive messages. */
-interface MessageReceiver {
+public interface MessageReceiver {
    enum AckReply {
     /** To be used for acking a message. */
     ACK,
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Publisher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Publisher.java
@@ -25,6 +25,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.pubsub.v1.PubsubMessage;
+import com.google.pubsub.v1.TopicName;
 import io.grpc.ManagedChannelBuilder;
 import java.io.IOException;
 import java.util.concurrent.ScheduledExecutorService;
@@ -191,8 +192,8 @@ public final class Builder {
     Optional<ScheduledExecutorService> executor = Optional.absent();
 
     /** Constructs a new {@link Builder} using the given topic. */
-    public static Builder newBuilder(String topic) {
-      return new Builder(topic);
+    public static Builder newBuilder(TopicName topic) {
+      return new Builder(topic.toString());
     }
 
     Builder(String topic) {
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java
@@ -29,6 +29,7 @@
 import com.google.common.util.concurrent.AbstractService;
 import com.google.common.util.concurrent.Service;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.pubsub.v1.SubscriptionName;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
@@ -401,8 +402,8 @@ public static final class Builder {
      * @param receiver an implementation of {@link MessageReceiver} used to process the received
      *     messages
      */
-    public static Builder newBuilder(String subscription, MessageReceiver receiver) {
-      return new Builder(subscription, receiver);
+    public static Builder newBuilder(SubscriptionName subscription, MessageReceiver receiver) {
+      return new Builder(subscription.toString(), receiver);
     }
 
     Builder(String subscription, MessageReceiver receiver) {
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/PublisherImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/PublisherImplTest.java
@@ -32,6 +32,7 @@
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
 import com.google.pubsub.v1.PubsubMessage;
+import com.google.pubsub.v1.TopicName;
 import io.grpc.Status;
 import io.grpc.StatusException;
 import io.grpc.inprocess.InProcessChannelBuilder;
@@ -57,7 +58,7 @@
 @RunWith(JUnit4.class)
 public class PublisherImplTest {
 
-  private static final String TEST_TOPIC = "projects/test-project/topics/test-topic";
+  private static final TopicName TEST_TOPIC = TopicName.create("test-project", "test-topic");
 
   private static InProcessChannelBuilder testChannelBuilder;
 
@@ -360,7 +361,7 @@ public void testPublisherGetters() throws Exception {
             .build());
     Publisher publisher = builder.build();
 
-    assertEquals(TEST_TOPIC, publisher.getTopic());
+    assertEquals(TEST_TOPIC.toString(), publisher.getTopic());
     assertEquals(10, publisher.getMaxBundleBytes());
     assertEquals(new Duration(11), publisher.getMaxBundleDuration());
     assertEquals(12, publisher.getMaxBundleMessages());
@@ -372,7 +373,7 @@ public void testPublisherGetters() throws Exception {
   @Test
   public void testBuilderParametersAndDefaults() {
     Publisher.Builder builder = Publisher.Builder.newBuilder(TEST_TOPIC);
-    assertEquals(TEST_TOPIC, builder.topic);
+    assertEquals(TEST_TOPIC.toString(), builder.topic);
     assertEquals(Optional.absent(), builder.channelBuilder);
     assertEquals(Optional.absent(), builder.executor);
     assertFalse(builder.failOnFlowControlLimits);
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java
@@ -33,6 +33,7 @@
 import com.google.pubsub.v1.PullResponse;
 import com.google.pubsub.v1.ReceivedMessage;
 import com.google.pubsub.v1.StreamingPullResponse;
+import com.google.pubsub.v1.SubscriptionName;
 import io.grpc.Status;
 import io.grpc.StatusException;
 import io.grpc.StatusRuntimeException;
@@ -63,8 +64,8 @@
 @RunWith(Parameterized.class)
 public class SubscriberImplTest {
 
-  private static final String TEST_SUBSCRIPTION =
-      "projects/test-project/subscriptions/test-subscription";
+  private static final SubscriptionName TEST_SUBSCRIPTION =
+      SubscriptionName.create("test-project", "test-subscription");
 
   private static final PubsubMessage TEST_MESSAGE =
       PubsubMessage.newBuilder().setMessageId("1").build();
