Google Cloud Pub/Sub

Spring Cloud GCP provides an abstraction layer to publish to and subscribe from Google Cloud Pub/Sub topics and to create, list or delete Google Cloud Pub/Sub topics and subscriptions.

A Spring Boot starter is provided to auto-configure the various required Pub/Sub components.

Maven coordinates, using Spring Cloud GCP BOM:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
</dependency>

Gradle coordinates:

dependencies {
    compile group: 'org.springframework.cloud', name: 'spring-cloud-gcp-starter-pubsub'
}

This starter is also available from Spring Initializr through the GCP Messaging entry.

Sample

Sample applications for using the template and using a subscription-backed reactive stream are available.

PubSubOperations is an abstraction that allows Spring users to use Google Cloud Pub/Sub without depending on any Google Cloud Pub/Sub API semantics. It provides the common set of operations needed to interact with Google Cloud Pub/Sub. PubSubTemplate is the default implementation of PubSubOperations and it uses the Google Cloud Java Client for Pub/Sub to interact with Google Cloud Pub/Sub.

Publishing to a topic

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:

Map<String, String> headers = Collections.singletonMap("key1", "val1");
pubSubTemplate.publish(topicName, "message", headers).get();

By default, the SimplePubSubMessageConverter is used to convert payloads of type byte[], ByteString, ByteBuffer, and String to Pub/Sub messages.

Subscribing to a subscription

Google Cloud Pub/Sub allows many subscriptions to be associated to the same topic. PubSubTemplate allows you to listen to subscriptions via the subscribe() method. When listening to a subscription, messages will be pulled from Google Cloud Pub/Sub asynchronously and passed to a user provided message handler.

Example

Subscribe to a subscription with a message handler:

Subscriber subscriber = pubSubTemplate.subscribe(subscriptionName, (message) -> {
	logger.info("Message received from " + subscriptionName + " subscription: "
			+ message.getPubsubMessage().getData().toStringUtf8());
	message.ack();
});

Subscribe methods

PubSubTemplate provides the following subscribe methods:

subscribe(String subscription, Consumer<BasicAcknowledgeablePubsubMessage> messageConsumer)	asynchronously pulls messages and passes them to `messageConsumer`
subscribeAndConvert(String subscription, Consumer<ConvertedBasicAcknowledgeablePubsubMessage<T>> messageConsumer, Class<T> payloadType)	same as `pull`, but converts message payload to `payloadType` using the converter configured in the template

subscribe(String subscription, Consumer<BasicAcknowledgeablePubsubMessage> messageConsumer)

asynchronously pulls messages and passes them to messageConsumer

subscribeAndConvert(String subscription, Consumer<ConvertedBasicAcknowledgeablePubsubMessage<T>> messageConsumer, Class<T> payloadType)

same as pull, but converts message payload to payloadType using the converter configured in the template

Pulling messages from a subscription

Google Cloud Pub/Sub supports synchronous pulling of messages from a subscription. This is different from subscribing to a subscription, in the sense that subscribing is an asynchronous task.

Example

Pull up to 10 messages:

int maxMessages = 10;
boolean returnImmediately = false;
List<AcknowledgeablePubsubMessage> messages = pubSubTemplate.pull(subscriptionName, maxMessages,
		returnImmediately);

//acknowledge the messages
pubSubTemplate.ack(messages);

messages.forEach(message -> logger.info(message.getPubsubMessage().getData().toStringUtf8()));

Pull methods

PubsubTemplate provides the following pull methods:

pull(String subscription, Integer maxMessages, Boolean returnImmediately)	Pulls a number of messages from a subscription, allowing for the retry settings to be configured. Any messages received by `pull()` are not automatically acknowledged. See Acknowledging messages. If `returnImmediately` is `true`, the system will respond immediately even if it there are no messages available to return in the `Pull` response. Otherwise, the system may wait (for a bounded amount of time) until at least one message is available, rather than returning no messages.
pullAndAck	Works the same as the `pull` method and, additionally, acknowledges all received messages.
pullNext	Allows for a single message to be pulled and automatically acknowledged from a subscription.
pullAndConvert	Works the same as the `pull` method and, additionally, converts the Pub/Sub binary payload to an object of the desired type, using the converter configured in the template.

pull(String subscription, Integer maxMessages, Boolean returnImmediately)

Pulls a number of messages from a subscription, allowing for the retry settings to be configured. Any messages received by pull() are not automatically acknowledged. See Acknowledging messages.

If returnImmediately is true, the system will respond immediately even if it there are no messages available to return in the Pull response. Otherwise, the system may wait (for a bounded amount of time) until at least one message is available, rather than returning no messages.

pullAndAck

Works the same as the pull method and, additionally, acknowledges all received messages.

pullNext

Allows for a single message to be pulled and automatically acknowledged from a subscription.

pullAndConvert

Works the same as the pull method and, additionally, converts the Pub/Sub binary payload to an object of the desired type, using the converter configured in the template.

Acknowledging messages

There are two ways to acknowledge messages.

To acknowledge multiple messages at once, you can use the PubSubTemplate.ack() method. You can also use the PubSubTemplate.nack() for negatively acknowledging messages. Using these methods for acknowledging messages in batches is more efficient than acknowledging messages individually, but they require the collection of messages to be from the same project.
To acknowledge messages individually you can use the ack() or nack() method on each of them (to acknowledge or negatively acknowledge, correspondingly).

All ack(), nack(), and modifyAckDeadline() methods on messages as well as PubSubSubscriberTemplate are implemented asynchronously, returning a ListenableFuture<Void> to be able to process the asynchronous execution.

JSON support

For serialization and deserialization of POJOs using Jackson JSON, configure a PubSubMessageConverter bean, and the Spring Boot starter for GCP Pub/Sub will automatically wire it into the PubSubTemplate.

// Note: The ObjectMapper is used to convert Java POJOs to and from JSON.
// You will have to configure your own instance if you are unable to depend
// on the ObjectMapper provided by Spring Boot starters.
@Bean
public PubSubMessageConverter pubSubMessageConverter() {
	return new JacksonPubSubMessageConverter(new ObjectMapper());
}

Alternatively, you can set it directly by calling the setMessageConverter() method on the PubSubTemplate. Other implementations of the PubSubMessageConverter can also be configured in the same manner.

Assuming you have the following class defined:

static class TestUser {

	String username;

	String password;

	public String getUsername() {
		return this.username;
	}

	void setUsername(String username) {
		this.username = username;
	}

	public String getPassword() {
		return this.password;
	}

	void setPassword(String password) {
		this.password = password;
	}
}

You can serialize objects to JSON on publish automatically:

TestUser user = new TestUser();
user.setUsername("John");
user.setPassword("password");
pubSubTemplate.publish(topicName, user);

And thatt’s how you convert messages to objects on pull:

int maxMessages = 1;
boolean returnImmediately = false;
List<ConvertedAcknowledgeablePubsubMessage<TestUser>> messages = pubSubTemplate.pullAndConvert(
		subscriptionName, maxMessages, returnImmediately, TestUser.class);

ConvertedAcknowledgeablePubsubMessage<TestUser> message = messages.get(0);

//acknowledge the message
message.ack();

TestUser receivedTestUser = message.getPayload();

Please refer to our Pub/Sub JSON Payload Sample App as a reference for using this functionality.

Reactive Stream Subscription

It is also possible to acquire a reactive stream backed by a subscription. To do so, a Project Reactor dependency (io.projectreactor:reactor-core) must be added to the project. The combination of the Pub/Sub starter and the Project Reactor dependencies will then make a PubSubReactiveFactory bean available, which can then be used to get a Publisher.

@Autowired
PubSubReactiveFactory reactiveFactory;

// ...

Flux<AcknowledgeablePubsubMessage> flux
				= reactiveFactory.createPolledFlux("exampleSubscription", 1000);

The Flux then represents an infinite stream of GCP Pub/Sub messages coming in through the specified subscription. For unlimited demand, the Pub/Sub subscription will be polled regularly, at intervals determined by pollingPeriodMs parameter passed in when creating the Flux. For bounded demand, the pollingPeriodMs parameter is unused. Instead, as many messages as possible (up to the requested number) are delivered immediately, with the remaining messages delivered as they become available.

The full range of Project Reactor operations can be applied to the stream. For example, if you only want to fetch 5 messages, you can use limitRequest operation to turn the infinite stream into a finite one:

Flux<AcknowledgeablePubsubMessage> fiveMessageFlux = flux.limitRequest(5);

Acknowledging messages flowing through the Flux should be manually acknowledged.

flux.doOnNext(AcknowledgeablePubsubMessage::ack);

Pub/Sub management

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.

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.

Creating a topic

PubSubAdmin implements a method to create topics:

public Topic createTopic(String topicName)

Here is an example of how to create a Google Cloud Pub/Sub topic:

public void newTopic() {
    pubSubAdmin.createTopic("topicName");
}

Deleting a topic

PubSubAdmin implements a method to delete topics:

public void deleteTopic(String topicName)

Here is an example of how to delete a Google Cloud Pub/Sub topic:

public void deleteTopic() {
    pubSubAdmin.deleteTopic("topicName");
}

Listing topics

PubSubAdmin implements a method to list topics:

public List<Topic> listTopics

Here is an example of how to list every Google Cloud Pub/Sub topic name in a project:

List<String> topics = pubSubAdmin
		.listTopics()
		.stream()
		.map(Topic::getName)
		.collect(Collectors.toList());

Creating a subscription

PubSubAdmin implements a method to create subscriptions to existing topics:

public Subscription createSubscription(String subscriptionName, String topicName, Integer ackDeadline, String pushEndpoint)

Here is an example of how to create a Google Cloud Pub/Sub subscription:

public void newSubscription() {
    pubSubAdmin.createSubscription("subscriptionName", "topicName", 10, “https://my.endpoint/push”);
}

Alternative methods with default settings are provided for ease of use. The default value for ackDeadline is 10 seconds. If pushEndpoint isn’t specified, the subscription uses message pulling, instead.

public Subscription createSubscription(String subscriptionName, String topicName)

public Subscription createSubscription(String subscriptionName, String topicName, Integer ackDeadline)

public Subscription createSubscription(String subscriptionName, String topicName, String pushEndpoint)

Deleting a subscription

PubSubAdmin implements a method to delete subscriptions:

public void deleteSubscription(String subscriptionName)

Here is an example of how to delete a Google Cloud Pub/Sub subscription:

public void deleteSubscription() {
    pubSubAdmin.deleteSubscription("subscriptionName");
}

Listing subscriptions

PubSubAdmin implements a method to list subscriptions:

public List<Subscription> listSubscriptions()

Here is an example of how to list every subscription name in a project:

List<String> subscriptions = pubSubAdmin
		.listSubscriptions()
		.stream()
		.map(Subscription::getName)
		.collect(Collectors.toList());

Configuration

The Spring Boot starter for Google Cloud Pub/Sub provides the following configuration options:

Name

Description

Required

Default value

spring.cloud.gcp.pubsub.enabled

Enables or disables Pub/Sub auto-configuration

true

spring.cloud.gcp.pubsub.project-id

GCP project ID where the Google Cloud Pub/Sub API is hosted, if different from the one in the Spring Cloud GCP Core Module

spring.cloud.gcp.pubsub.credentials.location

OAuth2 credentials for authenticating with the Google Cloud Pub/Sub API, if different from the ones in the Spring Cloud GCP Core Module

spring.cloud.gcp.pubsub.credentials.encoded-key

Base64-encoded contents of OAuth2 account private key for authenticating with the Google Cloud Pub/Sub API, if different from the ones in the Spring Cloud GCP Core Module

spring.cloud.gcp.pubsub.credentials.scopes

OAuth2 scope for Spring Cloud GCP Pub/Sub credentials

https://www.googleapis.com/auth/pubsub

spring.cloud.gcp.pubsub.keepAliveIntervalMinutes

Determines frequency of keepalive gRPC ping

5 minutes

spring.cloud.gcp.pubsub.subscriber.parallel-pull-count

The number of pull workers

The available number of processors

spring.cloud.gcp.pubsub.subscriber.max-ack-extension-period

The maximum period a message ack deadline will be extended, in seconds

spring.cloud.gcp.pubsub.subscriber.pull-endpoint

The endpoint for synchronous pulling messages

pubsub.googleapis.com:443

spring.cloud.gcp.pubsub.[subscriber,publisher].executor-threads

Number of threads used by Subscriber instances created by SubscriberFactory

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.total-timeout-seconds

TotalTimeout has ultimate control over how long the logic should keep trying the remote call until it gives up completely. The higher the total timeout, the more retries can be attempted.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.initial-retry-delay-second

InitialRetryDelay controls the delay before the first retry. Subsequent retries will use this value adjusted according to the RetryDelayMultiplier.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.retry-delay-multiplier

RetryDelayMultiplier controls the change in retry delay. The retry delay of the previous call is multiplied by the RetryDelayMultiplier to calculate the retry delay for the next call.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.max-retry-delay-seconds

MaxRetryDelay puts a limit on the value of the retry delay, so that the RetryDelayMultiplier can’t increase the retry delay higher than this amount.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.max-attempts

MaxAttempts defines the maximum number of attempts to perform. If this value is greater than 0, and the number of attempts reaches this limit, the logic will give up retrying even if the total retry time is still lower than TotalTimeout.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.jittered

Jitter determines if the delay time should be randomized.

true

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.initial-rpc-timeout-seconds

InitialRpcTimeout controls the timeout for the initial RPC. Subsequent calls will use this value adjusted according to the RpcTimeoutMultiplier.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.rpc-timeout-multiplier

RpcTimeoutMultiplier controls the change in RPC timeout. The timeout of the previous call is multiplied by the RpcTimeoutMultiplier to calculate the timeout for the next call.

spring.cloud.gcp.pubsub.[subscriber,publisher].retry.max-rpc-timeout-seconds

MaxRpcTimeout puts a limit on the value of the RPC timeout, so that the RpcTimeoutMultiplier can’t increase the RPC timeout higher than this amount.

spring.cloud.gcp.pubsub.[subscriber,publisher.batching].flow-control.max-outstanding-element-count

Maximum number of outstanding elements to keep in memory before enforcing flow control.

unlimited

spring.cloud.gcp.pubsub.[subscriber,publisher.batching].flow-control.max-outstanding-request-bytes

Maximum number of outstanding bytes to keep in memory before enforcing flow control.

unlimited

spring.cloud.gcp.pubsub.[subscriber,publisher.batching].flow-control.limit-exceeded-behavior

The behavior when the specified limits are exceeded.

Block

spring.cloud.gcp.pubsub.publisher.batching.element-count-threshold

The element count threshold to use for batching.

unset (threshold does not apply)

spring.cloud.gcp.pubsub.publisher.batching.request-byte-threshold

The request byte threshold to use for batching.

unset (threshold does not apply)

spring.cloud.gcp.pubsub.publisher.batching.delay-threshold-seconds

The delay threshold to use for batching. After this amount of time has elapsed (counting from the first element added), the elements will be wrapped up in a batch and sent.

unset (threshold does not apply)

spring.cloud.gcp.pubsub.publisher.batching.enabled

Enables batching.

false