Receive Pub/Sub Messages to Your Spring Application
Time to dive into GCP Pub/Sub.
CORE
·
Join the DZone community and get the full member experience.
Join For FreePub/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
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<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.
package com.gkatzioura.pubsub.example;
import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.stereotype.Component;
@Component
public 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.
@Bean
public MessageChannel pubsubInputChannel() {
return new DirectChannel();
}Then add the inbound channel adapter. The ack mode will be set to manual.
@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 acknowledgments are handled is up to the developer. If an exception occurs on that block, it will be caught and send on an error stream. Therefore, messages will continue to get pulled.
@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:
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;
@Configuration
public 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.
package com.gkatzioura.pubsub.example;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class UpdatesController {
private LatestUpdates latestUpdates;
public UpdatesController(LatestUpdates latestUpdates) {
this.latestUpdates = latestUpdates;
}
@GetMapping("/update")
public String getLatestUpdate() {
return latestUpdates.fetch();
}
}The next step is to define an application for Spring
package com.gkatzioura.pubsub.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public 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
spring.cloud.gcp.pubsub.enabled=trueThis will fall back to your Local GCP configuration and will identify your credentials as well as the project pointing at them.
That’s it! To summarise, we achieved to pull messages from Pub/Sub and expose them on an endpoint.
Published at DZone with permission of Emmanouil Gkatziouras. See the original article here.
Opinions expressed by DZone contributors are their own.
Comments