Pub/Sub is a messaging solution provided by GCP
Before we dive into the actual configuration we need to be aware that Spring Cloud for GCP is now managed by the Google Cloud Team. Therefore the latest code can be found here.
Our application will receive messages from Pub/Sub and expose them using an endpoint.
Let’s go for the imports first
01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | <?xml version="1.0" encoding="UTF-8"?> <modelVersion>4.0.0</modelVersion> <groupId>com.gkatzioura</groupId> <artifactId>spring-cloud-pubsub-example</artifactId> <version>1.0-SNAPSHOT</version> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.4.1</version> <relativePath/> </parent> <properties> <maven.compiler.source>11</maven.compiler.source> <maven.compiler.target>11</maven.compiler.target> </properties> <dependencyManagement> <dependencies> <dependency> <groupId>com.google.cloud</groupId> <artifactId>spring-cloud-gcp-dependencies</artifactId> <version>2.0.4</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>com.google.cloud</groupId> <artifactId>spring-cloud-gcp-pubsub</artifactId> </dependency> <dependency> <groupId>com.google.cloud</groupId> <artifactId>spring-cloud-gcp-autoconfigure</artifactId> </dependency> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-core</artifactId> </dependency> </dependencies></project> |
Quick note: with a few tweaks you can use the PubSub emulator available from the Google Cloud Team.
The first class will contain the Pub/Sub messages received. It will be a queue containing a limited number of messages.
01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 | package com.gkatzioura.pubsub.example;import java.util.concurrent.LinkedBlockingQueue;import org.springframework.stereotype.Component;@Componentpublic class LatestUpdates { LinkedBlockingQueue<String> boundedQueue = new LinkedBlockingQueue<>(100); public void addUpdate(String update) { boundedQueue.add(update); } public String fetch() { return boundedQueue.poll(); }} |
The Pub/Sub configuration will initiate the listener, plus shall use spring integration.
We define a message channel.
1 2 3 4 | @Bean public MessageChannel pubsubInputChannel() { return new DirectChannel(); } |
Then add the inbound channel adapter The ack mode will be set to manual.
01 02 03 04 05 06 07 08 09 10 11 | @Bean public PubSubInboundChannelAdapter messageChannelAdapter( @Qualifier("pubsubInputChannel") MessageChannel inputChannel, PubSubTemplate pubSubTemplate) { PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate, "your-subscription"); adapter.setOutputChannel(inputChannel); adapter.setAckMode(AckMode.MANUAL); adapter.setPayloadType(String.class); return adapter; } |
Then we add a listener method. The way acknowledgements are handled is up to the developer. If a exception occurs on that block it will be caught and send on an error stream. Therefore messages will continue to get pulled.
1 2 3 4 5 6 | @ServiceActivator(inputChannel = "pubsubInputChannel") public void messageReceiver(String payload, @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) BasicAcknowledgeablePubsubMessage message) { latestUpdates.addUpdate(message.getPubsubMessage().getData().toStringUtf8()); message.ack(); } |
The entire Pub/Sub configuration
01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | package com.gkatzioura.pubsub.example;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.integration.annotation.ServiceActivator;import org.springframework.integration.channel.DirectChannel;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.handler.annotation.Header;import com.google.cloud.spring.pubsub.core.PubSubTemplate;import com.google.cloud.spring.pubsub.integration.AckMode;import com.google.cloud.spring.pubsub.integration.inbound.PubSubInboundChannelAdapter;import com.google.cloud.spring.pubsub.support.BasicAcknowledgeablePubsubMessage;import com.google.cloud.spring.pubsub.support.GcpPubSubHeaders;@Configurationpublic class PubSubConfiguration { private final LatestUpdates latestUpdates; public PubSubConfiguration(LatestUpdates latestUpdates) { this.latestUpdates = latestUpdates; } @Bean public MessageChannel pubsubInputChannel() { return new DirectChannel(); } @Bean public PubSubInboundChannelAdapter messageChannelAdapter( @Qualifier("pubsubInputChannel") MessageChannel inputChannel, PubSubTemplate pubSubTemplate) { PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate, "your-subscription"); adapter.setOutputChannel(inputChannel); adapter.setAckMode(AckMode.MANUAL); adapter.setPayloadType(String.class); return adapter; } @ServiceActivator(inputChannel = "pubsubInputChannel") public void messageReceiver(String payload, @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) BasicAcknowledgeablePubsubMessage message) { latestUpdates.addUpdate(message.getPubsubMessage().getData().toStringUtf8()); message.ack(); }} |
The controller will just pull from the internal Queue.
01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 | package com.gkatzioura.pubsub.example;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;@RestControllerpublic class UpdatesController { private LatestUpdates latestUpdates; public UpdatesController(LatestUpdates latestUpdates) { this.latestUpdates = latestUpdates; } @GetMapping("/update") public String getLatestUpdate() { return latestUpdates.fetch(); }} |
Next step is to define an application for Spring
01 02 03 04 05 06 07 08 09 10 11 12 13 14 | package com.gkatzioura.pubsub.example;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class ExampleApplication { public static void main(String[] args) { SpringApplication.run(ExampleApplication.class, args); }} |
By running the application be aware that you need to have at least one env variable set
1 | spring.cloud.gcp.pubsub.enabled=true |
This will fallback to your Local GCP configuration and will identify your credentials as well as the project pointing at.
That’s it! To summarise, we achieved to pull messages from Pub/Sub and expose them on an endpoint.
Published on Java Code Geeks with permission by Emmanouil Gkatziouras, partner at our JCG program. See the original article here: Receive Pub/Sub messages to your Spring Application Opinions expressed by Java Code Geeks contributors are their own. |
