Add support for Google Cloud Pub/Sub

.travis.yml

@@ -9,7 +9,7 @@ before_install:
 - cp target/travis/settings.xml ~/.m2/settings.xml
 install: mvn install -DskipTests=true -Dgpg.skip=true
 script: 
-- travis_wait 30 utilities/verify.sh
+- travis_wait 60 utilities/verify.sh
 after_success:
 - utilities/after_success.sh
 env:

README.md

@@ -18,6 +18,7 @@ This client supports the following Google Cloud Platform services:
 -  [Google Cloud Compute] (#google-cloud-compute-alpha) (Alpha)
 -  [Google Cloud Datastore] (#google-cloud-datastore)
 -  [Google Cloud DNS] (#google-cloud-dns-alpha) (Alpha)
+-  [Google Cloud Pub/Sub] (#google-cloud-pubsub-alpha) (Alpha - Not working on App Engine Standard)
 -  [Google Cloud Resource Manager] (#google-cloud-resource-manager-alpha) (Alpha)
 -  [Google Cloud Storage] (#google-cloud-storage)
 
@@ -62,6 +63,8 @@ Example Applications
 - [`Flexible Environment/Datastore example`](https://github.com/GoogleCloudPlatform/java-docs-samples/tree/master/managed_vms/datastore) - A simple app that uses Cloud Datastore to list the last 10 IP addresses that visited your site.
   - Read about how to run the application [here](https://github.com/GoogleCloudPlatform/java-docs-samples/blob/master/managed_vms/README.md).
 - [`Flexible Environment/Storage example`](https://github.com/GoogleCloudPlatform/java-docs-samples/tree/master/managed_vms/cloudstorage) - An app that uploads files to a public Cloud Storage bucket on the App Engine Flexible Environment runtime.
+- [`PubSubExample`](./gcloud-java-examples/src/main/java/com/google/cloud/examples/pubsub/PubSubExample.java) - A simple command line interface providing some of Cloud Pub/Sub's functionality
+  - Read more about using this application on the [`PubSubExample` docs page](http://googlecloudplatform.github.io/gcloud-java/apidocs/?com/google/cloud/examples/pubsub/PubSubExample.html).
 - [`ResourceManagerExample`](./gcloud-java-examples/src/main/java/com/google/cloud/examples/resourcemanager/ResourceManagerExample.java) - A simple command line interface providing some of Cloud Resource Manager's functionality
   - Read more about using this application on the [`ResourceManagerExample` docs page](http://googlecloudplatform.github.io/gcloud-java/apidocs/?com/google/cloud/examples/resourcemanager/ResourceManagerExample.html).
 - [`SparkDemo`](https://github.com/GoogleCloudPlatform/java-docs-samples/blob/master/managed_vms/sparkjava) - An example of using `gcloud-java-datastore` from within the SparkJava and App Engine Flexible Environment frameworks.
@@ -368,6 +371,44 @@ ChangeRequestInfo changeRequest = changeBuilder.build();
 zone.applyChangeRequest(changeRequest);
 ```
 
+Google Cloud Pub/Sub (Alpha)
+----------------------
+
+- [API Documentation][pubsub-api]
+- [Official Documentation][cloud-pubsub-docs]
+
+#### Preview
+
+Here is a code snippet showing a simple usage example from within Compute Engine/App Engine
+Flexible. Note that you must [supply credentials](#authentication) and a project ID if running this
+snippet elsewhere. Complete source code can be found at
+[CreateSubscriptionAndPullMessages.java](./gcloud-java-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java).
+
+```java
+import com.google.cloud.pubsub.Message;
+import com.google.cloud.pubsub.PubSub;
+import com.google.cloud.pubsub.PubSub.MessageConsumer;
+import com.google.cloud.pubsub.PubSub.MessageProcessor;
+import com.google.cloud.pubsub.PubSubOptions;
+import com.google.cloud.pubsub.Subscription;
+import com.google.cloud.pubsub.SubscriptionInfo;
+
+try (PubSub pubsub = PubSubOptions.defaultInstance().service()) {
+  Subscription subscription =
+      pubsub.create(SubscriptionInfo.of("test-topic", "test-subscription"));
+  MessageProcessor callback = new MessageProcessor() {
+    @Override
+    public void process(Message message) throws Exception {
+      System.out.printf("Received message \"%s\"%n", message.payloadAsString());
+    }
+  };
+  // Create a message consumer and pull messages (for 60 seconds)
+  try (MessageConsumer consumer = subscription.pullAsync(callback)) {
+    Thread.sleep(60_000);
+  }
+}
+```
+
 Google Cloud Resource Manager (Alpha)
 ----------------------
 
@@ -513,6 +554,7 @@ Apache 2.0 - See [LICENSE] for more information.
 [cloud-dns-docs]: https://cloud.google.com/dns/docs
 [cloud-dns-activation]: https://console.cloud.google.com/start/api?id=dns
 
+[pubsub-api]: http://googlecloudplatform.github.io/gcloud-java/apidocs/index.html?com/google/cloud/pubsub/package-summary.html
 [cloud-pubsub]: https://cloud.google.com/pubsub/
 [cloud-pubsub-docs]: https://cloud.google.com/pubsub/docs
 

TESTING.md

@@ -174,5 +174,46 @@ Here is an example that clears the dataset created in Step 3.
   RemoteBigQueryHelper.forceDelete(bigquery, dataset);
   ```
 
+### Testing code that uses Pub/Sub
+
+#### On your machine
+
+You can test against a temporary local Pub/Sub by following these steps:
+
+1. Start the local Pub/Sub emulator before running your tests using `LocalPubSubHelper`'s `create`
+and `start` methods. This will bind a port for communication with the local Pub/Sub emulator.
+  ```java
+  LocalPubSubHelper helper = LocalPubSubHelper.create();
+
+  helper.start(); // Starts the local Pub/Sub emulator in a separate process
+  ```
+
+2. Create and use a `PubSub` object with the options given by the `LocalPubSubHelper` instance. For
+example:
+  ```java
+  PubSub localPubsub = helper.options().service();
+  ```
+
+3. Run your tests.
+
+4. Stop the local Pub/Sub emulator by calling the `stop()` method, like so:
+  ```java
+  helper.stop();
+  ```
+
+#### On a remote machine
+
+You can test against a remote Pub/Sub emulator as well. To do this, set the `PubSubOptions` project
+endpoint to the hostname of the remote machine, like the example below.
+
+  ```java
+  PubSubOptions options = PubSubOptions.builder()
+      .projectId("my-project-id") // must match project ID specified on remote machine
+      .host("<hostname of machine>:<port>")
+      .authCredentials(AuthCredentials.noAuth())
+      .build();
+  PubSub localPubsub= options.service();
+  ```
+
 [cloud-platform-storage-authentication]:https://cloud.google.com/storage/docs/authentication?hl=en#service_accounts
 [create-service-account]:https://developers.google.com/identity/protocols/OAuth2ServiceAccount#creatinganaccount

gcloud-java-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryOptions.java

@@ -16,19 +16,19 @@
 
 package com.google.cloud.bigquery;
 
-import com.google.cloud.ServiceOptions;
+import com.google.cloud.HttpServiceOptions;
 import com.google.cloud.bigquery.spi.BigQueryRpc;
 import com.google.cloud.bigquery.spi.BigQueryRpcFactory;
 import com.google.cloud.bigquery.spi.DefaultBigQueryRpc;
 import com.google.common.collect.ImmutableSet;
 
 import java.util.Set;
 
-public class BigQueryOptions extends ServiceOptions<BigQuery, BigQueryRpc, BigQueryOptions> {
+public class BigQueryOptions extends HttpServiceOptions<BigQuery, BigQueryRpc, BigQueryOptions> {
 
   private static final String BIGQUERY_SCOPE = "https://www.googleapis.com/auth/bigquery";
   private static final Set<String> SCOPES = ImmutableSet.of(BIGQUERY_SCOPE);
-  private static final long serialVersionUID = -215981591481708043L;
+  private static final long serialVersionUID = -8592198255032667206L;
 
   public static class DefaultBigqueryFactory implements BigQueryFactory {
 
@@ -51,7 +51,7 @@ public BigQueryRpc create(BigQueryOptions options) {
   }
 
   public static class Builder extends
-      ServiceOptions.Builder<BigQuery, BigQueryRpc, BigQueryOptions, Builder> {
+      HttpServiceOptions.Builder<BigQuery, BigQueryRpc, BigQueryOptions, Builder> {
 
     private Builder() {
     }

gcloud-java-compute/src/main/java/com/google/cloud/compute/ComputeOptions.java

@@ -16,19 +16,19 @@
 
 package com.google.cloud.compute;
 
-import com.google.cloud.ServiceOptions;
+import com.google.cloud.HttpServiceOptions;
 import com.google.cloud.compute.spi.ComputeRpc;
 import com.google.cloud.compute.spi.ComputeRpcFactory;
 import com.google.cloud.compute.spi.DefaultComputeRpc;
 import com.google.common.collect.ImmutableSet;
 
 import java.util.Set;
 
-public class ComputeOptions extends ServiceOptions<Compute, ComputeRpc, ComputeOptions> {
+public class ComputeOptions extends HttpServiceOptions<Compute, ComputeRpc, ComputeOptions> {
 
   private static final String COMPUTE_SCOPE = "https://www.googleapis.com/auth/compute";
   private static final Set<String> SCOPES = ImmutableSet.of(COMPUTE_SCOPE);
-  private static final long serialVersionUID = 6509557711917342058L;
+  private static final long serialVersionUID = 5074781985597996770L;
 
   public static class DefaultComputeFactory implements ComputeFactory {
 
@@ -51,7 +51,7 @@ public ComputeRpc create(ComputeOptions options) {
   }
 
   public static class Builder extends
-      ServiceOptions.Builder<Compute, ComputeRpc, ComputeOptions, Builder> {
+      HttpServiceOptions.Builder<Compute, ComputeRpc, ComputeOptions, Builder> {
 
     private Builder() {
     }

gcloud-java-core/pom.xml

@@ -98,5 +98,15 @@
       <version>3.4</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+      <version>3.0.0-beta-3</version>
+    </dependency>
+    <dependency>
+      <groupId>com.google.api</groupId>
+      <artifactId>gax</artifactId>
+      <version>0.0.13</version>
+    </dependency>
   </dependencies>
 </project>

gcloud-java-core/src/main/java/com/google/cloud/AsyncPage.java

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.util.concurrent.Future;
+
+/**
+ * Interface for asynchronously consuming Google Cloud paginated results.
+ *
+ * <p>Use {@code AsyncPage} to iterate through all values (also in next pages):
+ * <pre> {@code
+ * AsyncPage<T> page = ...; // get an AsyncPage<T> instance
+ * Iterator<T> iterator = page.iterateAll();
+ * while (iterator.hasNext()) {
+ *   T value = iterator.next();
+ *   // do something with value
+ * }}</pre>
+ *
+ * <p>Or handle pagination explicitly:
+ * <pre> {@code
+ * AsyncPage<T> page = ...; // get a AsyncPage<T> instance
+ * while (page != null) {
+ *   for (T value : page.values()) {
+ *     // do something with value
+ *   }
+ *   page = page.nextPageAsync().get();
+ * }}</pre>
+ *
+ * @param <T> the value type that the page holds
+ */
+public interface AsyncPage<T> extends Page<T> {
+
+  /**
+   * Returns a {@link Future} object for the next page. {@link Future#get()} returns {@code null} if
+   * the last page has been reached.
+   */
+  Future<AsyncPage<T>> nextPageAsync();
+}

gcloud-java-core/src/main/java/com/google/cloud/AsyncPageImpl.java

@@ -0,0 +1,83 @@
+/*
+ * 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;
+
+import com.google.common.base.Throwables;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.Uninterruptibles;
+
+import java.io.Serializable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Base implementation for asynchronously consuming Google Cloud paginated results.
+ *
+ * @param <T> the value type that the page holds
+ */
+public class AsyncPageImpl<T> extends PageImpl<T> implements AsyncPage<T> {
+
+  private static final long serialVersionUID = -6009473188630364906L;
+
+  private final NextPageFetcher<T> asyncPageFetcher;
+
+  /**
+   * Interface for asynchronously fetching the next page of results from the service.
+   *
+   * @param <T> the value type that the page holds
+   */
+  public interface NextPageFetcher<T> extends Serializable {
+    Future<AsyncPage<T>> nextPage();
+  }
+
+  private static class SyncNextPageFetcher<T> implements PageImpl.NextPageFetcher<T> {
+
+    private static final long serialVersionUID = -4124568632363525351L;
+
+    private final NextPageFetcher<T> asyncPageFetcher;
+
+    private SyncNextPageFetcher(NextPageFetcher<T> asyncPageFetcher) {
+      this.asyncPageFetcher = asyncPageFetcher;
+    }
+
+    @Override
+    public Page<T> nextPage() {
+      try {
+        return asyncPageFetcher != null
+            ? Uninterruptibles.getUninterruptibly(asyncPageFetcher.nextPage()) : null;
+      } catch (ExecutionException ex) {
+        throw Throwables.propagate(ex.getCause());
+      }
+    }
+  }
+
+  /**
+   * Creates an {@code AsyncPageImpl} object.
+   */
+  public AsyncPageImpl(NextPageFetcher<T> asyncPageFetcher, String cursor, Iterable<T> results) {
+    super(new SyncNextPageFetcher<T>(asyncPageFetcher), cursor, results);
+    this.asyncPageFetcher = asyncPageFetcher;
+  }
+
+  @Override
+  public Future<AsyncPage<T>> nextPageAsync() {
+    if (nextPageCursor() == null || asyncPageFetcher == null) {
+      return Futures.immediateCheckedFuture(null);
+    }
+    return asyncPageFetcher.nextPage();
+  }
+}

gcloud-java-core/src/main/java/com/google/cloud/BaseServiceException.java

@@ -18,6 +18,7 @@
 
 import com.google.api.client.googleapis.json.GoogleJsonError;
 import com.google.api.client.googleapis.json.GoogleJsonResponseException;
+import com.google.api.gax.grpc.ApiException;
 import com.google.common.base.MoreObjects;
 
 import java.io.IOException;
@@ -167,6 +168,16 @@ public BaseServiceException(int code, String message, String reason, boolean ide
     this.debugInfo = null;
   }
 
+  public BaseServiceException(ApiException apiException, boolean idempotent) {
+    super(apiException.getMessage(), apiException);
+    this.code = apiException.getStatusCode().value();
+    this.reason = apiException.getStatusCode().name();
+    this.idempotent = idempotent;
+    this.retryable = apiException.isRetryable();
+    this.location = null;
+    this.debugInfo = null;
+  }
+
   protected Set<Error> retryableErrors() {
     return Collections.emptySet();
   }