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
<?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 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.
@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();
}
}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=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. |
