PubsubTemplate publish to topics in other projects (#1678)

This change to the DefaultPublisherFactory allows the project id of the
topic to be overwitten directly from the topic specification in the
topic string.

This allows use-cases like
pubSubTemplate.publish("projects/other-project/topics/the-topic", "payload").

This change applied across the board to allow publishing, subscribing, creating,
and deleting topics using fully-qualified topic names.

Fixes #1678.

Author: meltsufin

docs/src/main/asciidoc/pubsub.adoc

@@ -39,6 +39,7 @@ It provides the common set of operations needed to interact with Google Cloud Pu
 
 `PubSubTemplate` provides asynchronous methods to publish messages to a Google Cloud Pub/Sub topic.
 The `publish()` method takes in a topic name to post the message to, a payload of a generic type and, optionally, a map with the message headers.
+The topic name could either be a canonical topic name within the current project, or the fully-qualified name referring to a topic in a different project using the `projects/<project_name>/topics/<topic_name>` format.
 
 Here is an example of how to publish a message to a Google Cloud Pub/Sub topic:
 
@@ -192,6 +193,8 @@ flux.doOnNext(AcknowledgeablePubsubMessage::ack);
 `PubSubAdmin` is the abstraction provided by Spring Cloud GCP to manage Google Cloud Pub/Sub resources.
 It allows for the creation, deletion and listing of topics and subscriptions.
 
+NOTE: Generally when referring to topics, you can either use the short canonical topic name within the current project, or the fully-qualified name referring to a topic in a different project using the `projects/<project_name>/topics/<topic_name>` format.
+
 `PubSubAdmin` depends on `GcpProjectIdProvider` and either a `CredentialsProvider` or a `TopicAdminClient` and a `SubscriptionAdminClient`.
 If given a `CredentialsProvider`, it creates a `TopicAdminClient` and a `SubscriptionAdminClient` with the Google Cloud Java Library for Pub/Sub default settings.
 The Spring Boot starter for GCP Pub/Sub auto-configures a `PubSubAdmin` object using the `GcpProjectIdProvider` and the `CredentialsProvider` auto-configured by the Spring Boot GCP Core starter.

spring-cloud-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/PubSubAdmin.java

@@ -30,16 +30,17 @@
 import com.google.cloud.pubsub.v1.TopicAdminSettings;
 import com.google.pubsub.v1.ProjectName;
 import com.google.pubsub.v1.ProjectSubscriptionName;
-import com.google.pubsub.v1.ProjectTopicName;
 import com.google.pubsub.v1.PushConfig;
 import com.google.pubsub.v1.Subscription;
 import com.google.pubsub.v1.Topic;
 
 import org.springframework.cloud.gcp.core.GcpProjectIdProvider;
+import org.springframework.cloud.gcp.pubsub.support.PubSubTopicUtils;
 import org.springframework.util.Assert;
 
 /**
- * Pub/Sub admin utility that creates new topics and subscriptions on Google Cloud Pub/Sub.
+ * Pub/Sub admin utility that creates new topics and subscriptions on Google Cloud
+ * Pub/Sub.
  *
  * @author João André Martins
  * @author Mike Eltsufin
@@ -96,26 +97,28 @@ public PubSubAdmin(GcpProjectIdProvider projectIdProvider, TopicAdminClient topi
 	/**
 	 * Create a new topic on Google Cloud Pub/Sub.
 	 *
-	 * @param topicName the name for the new topic
+	 * @param topicName the name for the new topic within the current project, or the
+	 * fully-qualified topic name in the projects/<project_name>/topics/<topic_name> format
 	 * @return the created topic
 	 */
 	public Topic createTopic(String topicName) {
 		Assert.hasText(topicName, "No topic name was specified.");
 
-		return this.topicAdminClient.createTopic(ProjectTopicName.of(this.projectId, topicName));
+		return this.topicAdminClient.createTopic(PubSubTopicUtils.toProjectTopicName(topicName, this.projectId));
 	}
 
 	/**
 	 * Get the configuration of a Google Cloud Pub/Sub topic.
 	 *
-	 * @param topicName canonical topic name, e.g., "topicName"
+	 * @param topicName canonical topic name, e.g., "topicName", or the fully-qualified topic name in the
+	 * "projects/<project_name>/topics/<topic_name>" format
 	 * @return topic configuration or {@code null} if topic doesn't exist
 	 */
 	public Topic getTopic(String topicName) {
 		Assert.hasText(topicName, "No topic name was specified.");
 
 		try {
-			return this.topicAdminClient.getTopic(ProjectTopicName.of(this.projectId, topicName));
+			return this.topicAdminClient.getTopic(PubSubTopicUtils.toProjectTopicName(topicName, this.projectId));
 		}
 		catch (ApiException aex) {
 			if (aex.getStatusCode().getCode() == StatusCode.Code.NOT_FOUND) {
@@ -129,12 +132,13 @@ public Topic getTopic(String topicName) {
 	/**
 	 * Delete a topic from Google Cloud Pub/Sub.
 	 *
-	 * @param topicName the name of the topic to be deleted
+	 * @param topicName canonical topic name, e.g., "topicName", or the fully-qualified topic
+	 * name in the "projects/<project_name>/topics/<topic_name>" format
 	 */
 	public void deleteTopic(String topicName) {
 		Assert.hasText(topicName, "No topic name was specified.");
 
-		this.topicAdminClient.deleteTopic(ProjectTopicName.of(this.projectId, topicName));
+		this.topicAdminClient.deleteTopic(PubSubTopicUtils.toProjectTopicName(topicName, this.projectId));
 	}
 
 	/**
@@ -167,8 +171,8 @@ public Subscription createSubscription(String subscriptionName, String topicName
 	 *
 	 * @param subscriptionName the name of the new subscription
 	 * @param topicName the name of the topic being subscribed to
-	 * @param ackDeadline deadline in seconds before a message is resent, must be between 10 and 600 seconds.
-	 *                    If not provided, set to default of 10 seconds
+	 * @param ackDeadline deadline in seconds before a message is resent, must be between 10
+	 * and 600 seconds. If not provided, set to default of 10 seconds
 	 * @return the created subscription
 	 */
 	public Subscription createSubscription(String subscriptionName, String topicName,
@@ -194,11 +198,12 @@ public Subscription createSubscription(String subscriptionName, String topicName
 	 * Create a new subscription on Google Cloud Pub/Sub.
 	 *
 	 * @param subscriptionName the name of the new subscription
-	 * @param topicName the name of the topic being subscribed to
-	 * @param ackDeadline deadline in seconds before a message is resent, must be between 10 and 600 seconds.
-	 *                    If not provided, set to default of 10 seconds
-	 * @param pushEndpoint the URL of the service receiving the push messages. If not provided, uses
-	 *                     message pulling by default
+	 * @param topicName canonical topic name, e.g., "topicName", or the fully-qualified topic name in the
+	 * "projects/<project_name>/topics/<topic_name>" format
+	 * @param ackDeadline deadline in seconds before a message is resent, must be between 10
+	 * and 600 seconds. If not provided, set to default of 10 seconds
+	 * @param pushEndpoint the URL of the service receiving the push messages. If not
+	 * provided, uses message pulling by default
 	 * @return the created subscription
 	 */
 	public Subscription createSubscription(String subscriptionName, String topicName,
@@ -219,7 +224,7 @@ public Subscription createSubscription(String subscriptionName, String topicName
 
 		return this.subscriptionAdminClient.createSubscription(
 				ProjectSubscriptionName.of(this.projectId, subscriptionName),
-				ProjectTopicName.of(this.projectId, topicName),
+				PubSubTopicUtils.toProjectTopicName(topicName, this.projectId),
 				pushConfigBuilder.build(),
 				finalAckDeadline);
 	}

spring-cloud-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/support/DefaultPublisherFactory.java

@@ -27,7 +27,6 @@
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.cloud.pubsub.v1.Publisher;
-import com.google.pubsub.v1.ProjectTopicName;
 
 import org.springframework.cloud.gcp.core.GcpProjectIdProvider;
 import org.springframework.cloud.gcp.pubsub.core.PubSubException;
@@ -65,7 +64,7 @@ public class DefaultPublisherFactory implements PublisherFactory {
 	/**
 	 * Create {@link DefaultPublisherFactory} instance based on the provided {@link GcpProjectIdProvider}.
 	 * <p>The {@link GcpProjectIdProvider} must not be null, neither provide an empty {@code projectId}.
-	 * @param projectIdProvider provides the GCP project ID
+	 * @param projectIdProvider provides the default GCP project ID for selecting the topic
 	 */
 	public DefaultPublisherFactory(GcpProjectIdProvider projectIdProvider) {
 		Assert.notNull(projectIdProvider, "The project ID provider can't be null.");
@@ -129,7 +128,7 @@ public Publisher createPublisher(String topic) {
 		return this.publishers.computeIfAbsent(topic, (key) -> {
 			try {
 				Publisher.Builder publisherBuilder =
-						Publisher.newBuilder(ProjectTopicName.of(this.projectId, key));
+						Publisher.newBuilder(PubSubTopicUtils.toProjectTopicName(topic, this.projectId));
 
 				if (this.executorProvider != null) {
 					publisherBuilder.setExecutorProvider(this.executorProvider);

spring-cloud-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/support/PubSubTopicUtils.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2017-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.cloud.gcp.pubsub.support;
+
+import com.google.pubsub.v1.ProjectTopicName;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Various utility methods for dealing with Pub/Sub topics.
+ *
+ * @author Mike Eltsufin
+ * @since 1.2
+ */
+public final class PubSubTopicUtils {
+
+	private PubSubTopicUtils() {
+	}
+
+	/**
+	 * Creates a {@link ProjectTopicName} based on a topic name within a project or the
+	 * fully-qualified topic name. If the specified topic is in the
+	 * projects/<project_name>/topics/<topic_name> format, then the {@code projectId} is
+	 * ignored}
+	 * @param topic the topic name in the project or the fully-qualified project name
+	 * @param projectId the project ID to use if the topic is not a fully-qualified name
+	 * @return the Pub/Sub object representing the topic name
+	 */
+	public static ProjectTopicName toProjectTopicName(String topic, @Nullable String projectId) {
+		Assert.notNull(topic, "The topic can't be null.");
+
+		ProjectTopicName projectTopicName = null;
+
+		if (ProjectTopicName.isParsableFrom(topic)) {
+			// Fully-qualified topic name in the "projects/<project_name>/topics/<topic_name>" format
+			projectTopicName = ProjectTopicName.parse(topic);
+		}
+		else {
+			Assert.notNull(projectId, "The project ID can't be null when using canonical topic name.");
+			projectTopicName = ProjectTopicName.of(projectId, topic);
+		}
+
+		return projectTopicName;
+	}
+}

spring-cloud-gcp-pubsub/src/test/java/org/springframework/cloud/gcp/pubsub/support/PubSubTopicUtilsTests.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2017-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.cloud.gcp.pubsub.support;
+
+import com.google.pubsub.v1.ProjectTopicName;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for {@link PubSubTopicUtils}.
+ *
+ * @author Mike Eltsufin
+ */
+public class PubSubTopicUtilsTests {
+
+	@Test
+	public void testToProjectTopicName_canonical() {
+		String project = "projectA";
+		String topic = "topicA";
+		String fqn = "projects/" + project + "/topics/" + topic;
+
+		ProjectTopicName parsedProjectTopicName = PubSubTopicUtils.toProjectTopicName(topic, project);
+
+		assertThat(parsedProjectTopicName).isEqualTo(ProjectTopicName.of(project, topic));
+		assertThat(parsedProjectTopicName.toString()).isEqualTo(fqn);
+	}
+
+	@Test
+	public void testToProjectTopicName_no_topic() {
+		assertThatThrownBy(() -> PubSubTopicUtils.toProjectTopicName(null, "topicA"))
+				.isInstanceOf(IllegalArgumentException.class)
+				.hasMessage("The topic can't be null.");
+	}
+
+	@Test
+	public void testToProjectTopicName_canonical_no_project() {
+		assertThatThrownBy(() -> PubSubTopicUtils.toProjectTopicName("topicA", null))
+				.isInstanceOf(IllegalArgumentException.class)
+				.hasMessage("The project ID can't be null when using canonical topic name.");
+	}
+
+	@Test
+	public void testToProjectTopicName_fqn() {
+		String project = "projectA";
+		String topic = "topicA";
+		String fqn = "projects/" + project + "/topics/" + topic;
+
+		ProjectTopicName parsedProjectTopicName = PubSubTopicUtils.toProjectTopicName(fqn, project);
+
+		assertThat(parsedProjectTopicName).isEqualTo(ProjectTopicName.of(project, topic));
+		assertThat(parsedProjectTopicName.toString()).isEqualTo(fqn);
+	}
+
+	@Test
+	public void testToProjectTopicName_fqn_no_project() {
+		String project = "projectA";
+		String topic = "topicA";
+		String fqn = "projects/" + project + "/topics/" + topic;
+
+		ProjectTopicName parsedProjectTopicName = PubSubTopicUtils.toProjectTopicName(fqn, null);
+
+		assertThat(parsedProjectTopicName).isEqualTo(ProjectTopicName.of(project, topic));
+		assertThat(parsedProjectTopicName.toString()).isEqualTo(fqn);
+	}
+}