From daff2a4083a75f5b65e0a53bfcbe8c1dea8bb0d8 Mon Sep 17 00:00:00 2001
From: Marco Ziccardi <marco.ziccard@gmail.com>
Date: Fri, 16 Sep 2016 15:25:54 +0200
Subject: [PATCH] Add snippets to Subscription's javadoc, SubscriptionSnippets
 class and tests

---
 .../pubsub/snippets/SubscriptionSnippets.java | 317 ++++++++++++++++++
 .../snippets/ITSubscriptionSnippets.java      | 125 +++++++
 .../com/google/cloud/pubsub/Subscription.java | 180 +++++++++-
 3 files changed, 604 insertions(+), 18 deletions(-)
 create mode 100644 google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionSnippets.java
 create mode 100644 google-cloud-examples/src/test/java/com/google/cloud/examples/pubsub/snippets/ITSubscriptionSnippets.java

diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionSnippets.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionSnippets.java
new file mode 100644
index 000000000000..ad4959f03ee6
--- /dev/null
+++ b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/SubscriptionSnippets.java
@@ -0,0 +1,317 @@
+/*
+ * Copyright 2016 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.
+ */
+
+/*
+ * EDITING INSTRUCTIONS
+ * This file is referenced in Subscription's javadoc. Any change to this file should be reflected in
+ * Subscription's javadoc.
+ */
+
+package com.google.cloud.examples.pubsub.snippets;
+
+import com.google.cloud.Identity;
+import com.google.cloud.Policy;
+import com.google.cloud.Role;
+import com.google.cloud.pubsub.Message;
+import com.google.cloud.pubsub.PubSub.MessageConsumer;
+import com.google.cloud.pubsub.PubSub.MessageProcessor;
+import com.google.cloud.pubsub.PushConfig;
+import com.google.cloud.pubsub.ReceivedMessage;
+import com.google.cloud.pubsub.Subscription;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * This class contains a number of snippets for the {@link Subscription} class.
+ */
+public class SubscriptionSnippets {
+
+  private final Subscription subscription;
+
+  public SubscriptionSnippets(Subscription subscription) {
+    this.subscription = subscription;
+  }
+
+  /**
+   * Example of getting the subscription's latest information.
+   */
+  // [TARGET reload()]
+  public Subscription reload() {
+    // [START reload]
+    Subscription latestSubscription = subscription.reload();
+    if (latestSubscription == null) {
+      // the subscription was not found
+    }
+    // [END reload]
+    return latestSubscription;
+  }
+
+  /**
+   * Example of asynchronously getting the subscription's latest information.
+   */
+  // [TARGET reloadAsync()]
+  public Subscription reloadAsync() throws ExecutionException, InterruptedException {
+    // [START reloadAsync]
+    Future<Subscription> future = subscription.reloadAsync();
+    // ...
+    Subscription latestSubscription = future.get();
+    if (latestSubscription == null) {
+      // the subscription was not found
+    }
+    // [END reloadAsync]
+    return latestSubscription;
+  }
+
+  /**
+   * Example of deleting the subscription.
+   */
+  // [TARGET delete()]
+  public boolean delete() {
+    // [START delete]
+    boolean deleted = subscription.delete();
+    if (deleted) {
+      // the subscription was deleted
+    } else {
+      // the subscription was not found
+    }
+    // [END delete]
+    return deleted;
+  }
+
+  /**
+   * Example of asynchronously deleting the subscription.
+   */
+  // [TARGET deleteAsync()]
+  public boolean deleteAsync() throws ExecutionException, InterruptedException {
+    // [START deleteAsync]
+    Future<Boolean> future = subscription.deleteAsync();
+    // ...
+    boolean deleted = future.get();
+    if (deleted) {
+      // the subscription was deleted
+    } else {
+      // the subscription was not found
+    }
+    // [END deleteAsync]
+    return deleted;
+  }
+
+  /**
+   * Example of replacing the push configuration of the subscription, setting the push endpoint.
+   */
+  // [TARGET replacePushConfig(PushConfig)]
+  // [VARIABLE "https://www.example.com/push"]
+  public void replacePushConfig(String endpoint) {
+    // [START replacePushConfig]
+    PushConfig pushConfig = PushConfig.of(endpoint);
+    subscription.replacePushConfig(pushConfig);
+    // [END replacePushConfig]
+  }
+
+  /**
+   * Example of replacing the push configuration of the subscription, making it a pull
+   * subscription.
+   */
+  // [TARGET replacePushConfig(PushConfig)]
+  public void replacePushConfigToPull() {
+    // [START replacePushConfigToPull]
+    subscription.replacePushConfig(null);
+    // [END replacePushConfigToPull]
+  }
+
+  /**
+   * Example of asynchronously replacing the push configuration of the subscription, setting the
+   * push endpoint.
+   */
+  // [TARGET replacePushConfigAsync(PushConfig)]
+  // [VARIABLE "https://www.example.com/push"]
+  public void replacePushConfigAsync(String endpoint)
+      throws ExecutionException, InterruptedException {
+    // [START replacePushConfigAsync]
+    PushConfig pushConfig = PushConfig.of(endpoint);
+    Future<Void> future = subscription.replacePushConfigAsync(pushConfig);
+    // ...
+    future.get();
+    // [END replacePushConfigAsync]
+  }
+
+  /**
+   * Example of asynchronously replacing the push configuration of the subscription, making it a
+   * pull subscription.
+   */
+  // [TARGET replacePushConfigAsync(PushConfig)]
+  public void replacePushConfigToPullAsync()
+      throws ExecutionException, InterruptedException {
+    // [START replacePushConfigToPullAsync]
+    Future<Void> future = subscription.replacePushConfigAsync(null);
+    // ...
+    future.get();
+    // [END replacePushConfigToPullAsync]
+  }
+
+  /**
+   * Example of pulling a maximum number of messages from the subscription.
+   */
+  // [TARGET pull(int)]
+  public void pull() {
+    // [START pull]
+    Iterator<ReceivedMessage> messages = subscription.pull(100);
+    // Ack deadline is renewed until the message is consumed
+    while (messages.hasNext()) {
+      ReceivedMessage message = messages.next();
+      // do something with message and ack/nack it
+      message.ack(); // or message.nack()
+    }
+    // [END pull]
+  }
+
+  /**
+   * Example of asynchronously pulling a maximum number of messages from the subscription.
+   */
+  // [TARGET pullAsync(int)]
+  public void pullAsync() throws ExecutionException, InterruptedException {
+    // [START pullAsync]
+    Future<Iterator<ReceivedMessage>> future = subscription.pullAsync(100);
+    // ...
+    Iterator<ReceivedMessage> messages = future.get();
+    // Ack deadline is renewed until the message is consumed
+    while (messages.hasNext()) {
+      ReceivedMessage message = messages.next();
+      // do something with message and ack/nack it
+      message.ack(); // or message.nack()
+    }
+    // [END pullAsync]
+  }
+
+  /**
+   * Example of continuously pulling messages from the subscription.
+   */
+  // [TARGET pullAsync(MessageProcessor, PullOption...)]
+  // [VARIABLE "my_subscription_name"]
+  public void pullWithMessageConsumer(String subscriptionName) throws Exception {
+    // [START pullWithMessageConsumer]
+    MessageProcessor callback = new MessageProcessor() {
+      @Override
+      public void process(Message message) throws Exception {
+        // Ack deadline is renewed until this method returns
+        // Message is acked if this method returns successfully
+        // Message is nacked if this method throws an exception
+      }
+    };
+    MessageConsumer consumer = subscription.pullAsync(callback);
+    // ...
+    // Stop pulling
+    consumer.close();
+    // [END pullWithMessageConsumer]
+  }
+
+  /**
+   * Example of getting the subscription's policy.
+   */
+  // [TARGET getPolicy()]
+  public Policy getPolicy() {
+    // [START getPolicy]
+    Policy policy = subscription.getPolicy();
+    if (policy == null) {
+      // subscription was not found
+    }
+    // [END getPolicy]
+    return policy;
+  }
+
+  /**
+   * Example of asynchronously getting the subscription's policy.
+   */
+  // [TARGET getPolicyAsync()]
+  public Policy getPolicyAsync() throws ExecutionException, InterruptedException {
+    // [START getPolicyAsync]
+    Future<Policy> future = subscription.getPolicyAsync();
+    // ...
+    Policy policy = future.get();
+    if (policy == null) {
+      // subscription was not found
+    }
+    // [END getPolicyAsync]
+    return policy;
+  }
+
+  /**
+   * Example of replacing the subscription's policy.
+   */
+  // [TARGET replacePolicy(Policy)]
+  public Policy replacePolicy() {
+    // [START replacePolicy]
+    Policy policy = subscription.getPolicy();
+    Policy updatedPolicy = policy.toBuilder()
+        .addIdentity(Role.viewer(), Identity.allAuthenticatedUsers())
+        .build();
+    updatedPolicy = subscription.replacePolicy(updatedPolicy);
+    // [END replacePolicy]
+    return updatedPolicy;
+  }
+
+  /**
+   * Example of asynchronously replacing the subscription's policy.
+   */
+  // [TARGET replacePolicyAsync(Policy)]
+  public Policy replacePolicyAsync()
+      throws ExecutionException, InterruptedException {
+    // [START replacePolicyAsync]
+    Policy policy = subscription.getPolicy();
+    Policy updatedPolicy = policy.toBuilder()
+        .addIdentity(Role.viewer(), Identity.allAuthenticatedUsers())
+        .build();
+    Future<Policy> future = subscription.replacePolicyAsync(updatedPolicy);
+    // ...
+    updatedPolicy = future.get();
+    // [END replacePolicyAsync]
+    return updatedPolicy;
+  }
+
+  /**
+   * Example of testing whether the caller has the provided permissions on the subscription.
+   */
+  // [TARGET testPermissions(List)]
+  public List<Boolean> testPermissions() {
+    // [START testPermissions]
+    List<String> permissions = new LinkedList<>();
+    permissions.add("pubsub.subscriptions.get");
+    List<Boolean> testedPermissions = subscription.testPermissions(permissions);
+    // [END testPermissions]
+    return testedPermissions;
+  }
+
+  /**
+   * Example of asynchronously testing whether the caller has the provided permissions on the
+   * subscription.
+   */
+  // [TARGET testPermissionsAsync(List)]
+  public List<Boolean> testPermissionsAsync()
+      throws ExecutionException, InterruptedException {
+    // [START testPermissionsAsync]
+    List<String> permissions = new LinkedList<>();
+    permissions.add("pubsub.subscriptions.get");
+    Future<List<Boolean>> future = subscription.testPermissionsAsync(permissions);
+    // ...
+    List<Boolean> testedPermissions = future.get();
+    // [END testPermissionsAsync]
+    return testedPermissions;
+  }
+}
diff --git a/google-cloud-examples/src/test/java/com/google/cloud/examples/pubsub/snippets/ITSubscriptionSnippets.java b/google-cloud-examples/src/test/java/com/google/cloud/examples/pubsub/snippets/ITSubscriptionSnippets.java
new file mode 100644
index 000000000000..7e6f89d4074f
--- /dev/null
+++ b/google-cloud-examples/src/test/java/com/google/cloud/examples/pubsub/snippets/ITSubscriptionSnippets.java
@@ -0,0 +1,125 @@
+/*
+ * Copyright 2016 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.Identity;
+import com.google.cloud.Policy;
+import com.google.cloud.Role;
+import com.google.cloud.pubsub.Message;
+import com.google.cloud.pubsub.PubSub;
+import com.google.cloud.pubsub.PubSubOptions;
+import com.google.cloud.pubsub.ReceivedMessage;
+import com.google.cloud.pubsub.Subscription;
+import com.google.cloud.pubsub.SubscriptionInfo;
+import com.google.cloud.pubsub.Topic;
+import com.google.cloud.pubsub.TopicInfo;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+public class ITSubscriptionSnippets {
+
+  private static final String TOPIC =
+      "it-subscription-snippets-topic-" + UUID.randomUUID().toString();
+  private static final String SUBSCRIPTION =
+      "it-subscription-snippets-subscription-" + UUID.randomUUID().toString();
+  private static final Message MESSAGE1 = Message.of("message1");
+  private static final Message MESSAGE2 = Message.of("message2");
+
+  private static PubSub pubsub;
+  private static Topic topic;
+  private static Subscription subscription;
+
+  @BeforeClass
+  public static void beforeClass() {
+    pubsub = PubSubOptions.defaultInstance().service();
+    topic = pubsub.create(TopicInfo.of(TOPIC));
+    subscription = pubsub.create(SubscriptionInfo.of(TOPIC, SUBSCRIPTION));
+  }
+
+  @AfterClass
+  public static void afterClass() throws Exception {
+    if (pubsub != null) {
+      topic.delete();
+      subscription.delete();
+      pubsub.close();
+    }
+  }
+
+  @Test
+  public void testPushConfig() throws ExecutionException, InterruptedException {
+    SubscriptionSnippets subscriptionSnippets = new SubscriptionSnippets(subscription);
+    String endpoint = "https://" + pubsub.options().projectId() + ".appspot.com/push";
+    subscriptionSnippets.replacePushConfig(endpoint);
+    Subscription updatedSubscription = pubsub.getSubscription(SUBSCRIPTION);
+    assertEquals(endpoint, updatedSubscription.pushConfig().endpoint());
+    subscriptionSnippets.replacePushConfigToPull();
+    updatedSubscription = pubsub.getSubscription(SUBSCRIPTION);
+    assertNull(updatedSubscription.pushConfig());
+    subscriptionSnippets.replacePushConfigAsync(endpoint);
+    updatedSubscription = pubsub.getSubscription(SUBSCRIPTION);
+    assertEquals(endpoint, updatedSubscription.pushConfig().endpoint());
+    subscriptionSnippets.replacePushConfigToPullAsync();
+    updatedSubscription = pubsub.getSubscription(SUBSCRIPTION);
+    assertNull(updatedSubscription.pushConfig());
+  }
+
+  @Test
+  public void testPull() throws ExecutionException, InterruptedException {
+    SubscriptionSnippets subscriptionSnippets = new SubscriptionSnippets(subscription);
+    pubsub.publish(TOPIC, MESSAGE1, MESSAGE2);
+    subscriptionSnippets.pull();
+    // messages have been acked, we should pull nothing
+    Iterator<ReceivedMessage> iterator = pubsub.pull(SUBSCRIPTION, 2);
+    assertFalse(iterator.hasNext());
+    pubsub.publish(TOPIC, MESSAGE1, MESSAGE2);
+    subscriptionSnippets.pullAsync();
+    // messages have been acked, we should pull nothing
+    iterator = pubsub.pull(SUBSCRIPTION, 2);
+    assertFalse(iterator.hasNext());
+    subscriptionSnippets.pullAsync();
+  }
+
+  @Test
+  public void testPolicy() throws ExecutionException, InterruptedException {
+    SubscriptionSnippets subscriptionSnippets = new SubscriptionSnippets(subscription);
+    Policy policy = subscriptionSnippets.getPolicy();
+    assertNotNull(policy);
+    assertEquals(policy, subscriptionSnippets.getPolicyAsync());
+    policy = subscriptionSnippets.replacePolicy();
+    assertTrue(policy.bindings().get(Role.viewer()).contains(Identity.allAuthenticatedUsers()));
+    policy = subscription.replacePolicy(policy.toBuilder()
+        .removeIdentity(Role.viewer(), Identity.allAuthenticatedUsers())
+        .build());
+    assertFalse(policy.bindings().containsKey(Role.viewer()));
+    policy = subscriptionSnippets.replacePolicyAsync();
+    assertTrue(policy.bindings().get(Role.viewer()).contains(Identity.allAuthenticatedUsers()));
+    assertTrue(subscriptionSnippets.delete());
+    assertFalse(subscriptionSnippets.deleteAsync());
+  }
+}
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/Subscription.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/Subscription.java
index afb064c734f9..bbb94aafbe35 100644
--- a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/Subscription.java
+++ b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/Subscription.java
@@ -159,6 +159,16 @@ public PubSub pubSub() {
   /**
    * Deletes this subscription.
    *
+   * <p>Example of deleting the subscription.
+   * <pre> {@code
+   * boolean deleted = subscription.delete();
+   * if (deleted) {
+   *   // the subscription was deleted
+   * } else {
+   *   // the subscription was not found
+   * }
+   * }</pre>
+   *
    * @return {@code true} if the subscription was deleted, {@code false} if it was not found
    * @throws PubSubException upon failure
    */
@@ -170,6 +180,19 @@ public boolean delete() {
    * Sends a request for deleting this subscription. This method returns a {@code Future} object to
    * consume the result. {@link Future#get()} returns {@code true} if the subscription was deleted,
    * {@code false} if it was not found.
+   *
+   * <p>Example of asynchronously deleting the subscription.
+   * <pre> {@code
+   * Future<Boolean> future = subscription.deleteAsync();
+   * // ...
+   * boolean deleted = future.get();
+   * if (deleted) {
+   *   // the subscription was deleted
+   * } else {
+   *   // the subscription was not found
+   * }
+   * }</pre>
+   *
    */
   public Future<Boolean> deleteAsync() {
     return pubsub.deleteSubscriptionAsync(name());
@@ -179,6 +202,14 @@ public Future<Boolean> deleteAsync() {
    * Fetches current subscription's latest information. Returns {@code null} if the subscription
    * does not exist.
    *
+   * <p>Example of getting the subscription's latest information.
+   * <pre> {@code
+   * Subscription latestSubscription = subscription.reload();
+   * if (latestSubscription == null) {
+   *   // the subscription was not found
+   * }
+   * }</pre>
+   *
    * @return a {@code Subscription} object with latest information or {@code null} if not found
    * @throws PubSubException upon failure
    */
@@ -191,6 +222,16 @@ public Subscription reload() {
    * {@code Future} object to consume the result. {@link Future#get()} returns the requested
    * subscription or {@code null} if not found.
    *
+   * <p>Example of asynchronously getting the subscription's latest information.
+   * <pre> {@code
+   * Future<Subscription> future = subscription.reloadAsync();
+   * // ...
+   * Subscription latestSubscription = future.get();
+   * if (latestSubscription == null) {
+   *   // the subscription was not found
+   * }
+   * }</pre>
+   *
    * @return a {@code Subscription} object with latest information or {@code null} if not found
    * @throws PubSubException upon failure
    */
@@ -205,6 +246,19 @@ public Future<Subscription> reloadAsync() {
    * subscription. Messages will accumulate for delivery regardless of changes to the push
    * configuration.
    *
+   * <p>Example of replacing the push configuration of the subscription, setting the push endpoint.
+   * <pre> {@code
+   * String endpoint = "https://www.example.com/push";
+   * PushConfig pushConfig = PushConfig.of(endpoint);
+   * subscription.replacePushConfig(pushConfig);
+   * }</pre>
+   *
+   * <p>Example of replacing the push configuration of the subscription, making it a pull
+   * subscription.
+   * <pre> {@code
+   * subscription.replacePushConfig(null);
+   * }</pre>
+   *
    * @param pushConfig the new push configuration. Use {@code null} to unset it
    * @throws PubSubException upon failure, or if the subscription does not exist
    */
@@ -220,6 +274,24 @@ public void replacePushConfig(PushConfig pushConfig) {
    * to the push configuration. The method returns a {@code Future} object that can be used to wait
    * for the replace operation to be completed.
    *
+   * <p>Example of asynchronously replacing the push configuration of the subscription, setting the
+   * push endpoint.
+   * <pre> {@code
+   * String endpoint = "https://www.example.com/push";
+   * PushConfig pushConfig = PushConfig.of(endpoint);
+   * Future<Void> future = subscription.replacePushConfigAsync(pushConfig);
+   * // ...
+   * future.get();
+   * }</pre>
+   *
+   * <p>Example of asynchronously replacing the push configuration of the subscription, making it a
+   * pull subscription.
+   * <pre> {@code
+   * Future<Void> future = subscription.replacePushConfigAsync(null);
+   * // ...
+   * future.get();
+   * }</pre>
+   *
    * @param pushConfig the new push configuration. Use {@code null} to unset it
    * @return a {@code Future} to wait for the replace operation to be completed.
    */
@@ -234,16 +306,16 @@ public Future<Void> replacePushConfigAsync(PushConfig pushConfig) {
    * acknowledge deadline automatically renewed until they are explicitly consumed using
    * {@link Iterator#next()}.
    *
-   * <p>Example usage of synchronous message pulling:
+   * <p>Example of pulling a maximum number of messages from the subscription.
    * <pre> {@code
-   * Iterator<ReceivedMessage> messageIterator = pubsub.pull("subscription", 100);
-   * while (messageIterator.hasNext()) {
-   *   ReceivedMessage message = messageIterator.next();
-   *   // message's acknowledge deadline is no longer automatically renewed. If processing takes
-   *   // long pubsub.modifyAckDeadline(String, String, long, TimeUnit) can be used to extend it.
-   *   doSomething(message);
+   * Iterator<ReceivedMessage> messages = subscription.pull(100);
+   * // Ack deadline is renewed until the message is consumed
+   * while (messages.hasNext()) {
+   *   ReceivedMessage message = messages.next();
+   *   // do something with message and ack/nack it
    *   message.ack(); // or message.nack()
-   * }}</pre>
+   * }
+   * }</pre>
    *
    * @param maxMessages the maximum number of messages pulled by this method. This method can
    *     possibly return fewer messages.
@@ -260,18 +332,18 @@ public Iterator<ReceivedMessage> pull(int maxMessages) {
    * was processed by the Pub/Sub service (i.e. the system is not allowed to wait until at least one
    * message is available).
    *
-   * <p>Example usage of asynchronous message pulling:
+   * <p>Example of asynchronously pulling a maximum number of messages from the subscription.
    * <pre> {@code
-   * Future<Iterator<ReceivedMessage>> future = pubsub.pull("subscription", 100);
-   * // do something while the request gets processed
-   * Iterator<ReceivedMessage> messageIterator = future.get();
-   * while (messageIterator.hasNext()) {
-   *   ReceivedMessage message = messageIterator.next();
-   *   // message's acknowledge deadline is no longer automatically renewed. If processing takes
-   *   // long pubsub.modifyAckDeadline(String, String, long, TimeUnit) can be used to extend it.
-   *   doSomething(message);
+   * Future<Iterator<ReceivedMessage>> future = subscription.pullAsync(100);
+   * // ...
+   * Iterator<ReceivedMessage> messages = future.get();
+   * // Ack deadline is renewed until the message is consumed
+   * while (messages.hasNext()) {
+   *   ReceivedMessage message = messages.next();
+   *   // do something with message and ack/nack it
    *   message.ack(); // or message.nack()
-   * }}</pre>
+   * }
+   * }</pre>
    *
    * @param maxMessages the maximum number of messages pulled by this method. This method can
    *     possibly return fewer messages.
@@ -295,6 +367,23 @@ public Future<Iterator<ReceivedMessage>> pullAsync(int maxMessages) {
    * {@link PullOption#executorFactory(GrpcServiceOptions.ExecutorFactory)} can be used to provide
    * an executor to run message processor callbacks.
    *
+   * <p>Example of continuously pulling messages from the subscription.
+   * <pre> {@code
+   * String subscriptionName = "my_subscription_name";
+   * MessageProcessor callback = new MessageProcessor() {
+   *   @Override
+   *   public void process(Message message) throws Exception {
+   *     // Ack deadline is renewed until this method returns
+   *     // Message is acked if this method returns successfully
+   *     // Message is nacked if this method throws an exception
+   *   }
+   * };
+   * MessageConsumer consumer = subscription.pullAsync(callback);
+   * // ...
+   * // Stop pulling
+   * consumer.close();
+   * }</pre>
+   *
    * @param callback the callback to be executed on each message
    * @param options pulling options
    * @return a message consumer for the provided subscription and options
@@ -307,6 +396,14 @@ public MessageConsumer pullAsync(MessageProcessor callback, PullOption... option
    * Returns the IAM access control policy for this subscription. Returns {@code null} if the
    * subscription was not found.
    *
+   * <p>Example of getting the subscription's policy.
+   * <pre> {@code
+   * Policy policy = subscription.getPolicy();
+   * if (policy == null) {
+   *   // subscription was not found
+   * }
+   * }</pre>
+   *
    * @throws PubSubException upon failure
    */
   public Policy getPolicy() {
@@ -318,6 +415,16 @@ public Policy getPolicy() {
    * returns a {@code Future} object to consume the result. {@link Future#get()} returns the
    * requested policy or {@code null} if the subscription was not found.
    *
+   * <p>Example of asynchronously getting the subscription's policy.
+   * <pre> {@code
+   * Future<Policy> future = subscription.getPolicyAsync();
+   * // ...
+   * Policy policy = future.get();
+   * if (policy == null) {
+   *   // subscription was not found
+   * }
+   * }</pre>
+   *
    * @throws PubSubException upon failure
    */
   public Future<Policy> getPolicyAsync() {
@@ -338,6 +445,15 @@ public Future<Policy> getPolicyAsync() {
    * {@code PubSubException} is thrown, denoting that the server aborted update. If an etag is not
    * provided, the policy is overwritten blindly.
    *
+   * <p>Example of replacing the subscription's policy.
+   * <pre> {@code
+   * Policy policy = subscription.getPolicy();
+   * Policy updatedPolicy = policy.toBuilder()
+   *     .addIdentity(Role.viewer(), Identity.allAuthenticatedUsers())
+   *     .build();
+   * updatedPolicy = subscription.replacePolicy(updatedPolicy);
+   * }</pre>
+   *
    * @throws PubSubException upon failure
    */
   public Policy replacePolicy(Policy newPolicy) {
@@ -360,6 +476,17 @@ public Policy replacePolicy(Policy newPolicy) {
    * {@code PubSubException}, denoting that the server aborted update. If an etag is not provided,
    * the policy is overwritten blindly.
    *
+   * <p>Example of asynchronously replacing the subscription's policy.
+   * <pre> {@code
+   * Policy policy = subscription.getPolicy();
+   * Policy updatedPolicy = policy.toBuilder()
+   *     .addIdentity(Role.viewer(), Identity.allAuthenticatedUsers())
+   *     .build();
+   * Future<Policy> future = subscription.replacePolicyAsync(updatedPolicy);
+   * // ...
+   * updatedPolicy = future.get();
+   * }</pre>
+   *
    * @throws PubSubException upon failure
    */
   public Future<Policy> replacePolicyAsync(Policy newPolicy) {
@@ -373,6 +500,13 @@ public Future<Policy> replacePolicyAsync(Policy newPolicy) {
    * interface. For example, the Cloud Platform Console tests IAM permissions internally to
    * determine which UI should be available to the logged-in user.
    *
+   * <p>Example of testing whether the caller has the provided permissions on the subscription.
+   * <pre> {@code
+   * List<String> permissions = new LinkedList<>();
+   * permissions.add("pubsub.subscriptions.get");
+   * List<Boolean> testedPermissions = subscription.testPermissions(permissions);
+   * }</pre>
+   *
    * @return A list of booleans representing whether the caller has the permissions specified (in
    *     the order of the given permissions)
    * @throws PubSubException upon failure
@@ -391,6 +525,16 @@ public List<Boolean> testPermissions(List<String> permissions) {
    * such as a customized graphical user interface. For example, the Cloud Platform Console tests
    * IAM permissions internally to determine which UI should be available to the logged-in user.
    *
+   * <p>Example of asynchronously testing whether the caller has the provided permissions on the
+   * subscription.
+   * <pre> {@code
+   * List<String> permissions = new LinkedList<>();
+   * permissions.add("pubsub.subscriptions.get");
+   * Future<List<Boolean>> future = subscription.testPermissionsAsync(permissions);
+   * // ...
+   * List<Boolean> testedPermissions = future.get();
+   * }</pre>
+   *
    * @return A {@code Future} object to consume the result. {@link Future#get()} returns a list of
    *     booleans representing whether the caller has the permissions specified (in the order of the
    *     given permissions)