Receive Pub/Sub Messages to Your Spring Application

Time to dive into GCP Pub/Sub.

Emmanouil Gkatziouras

CORE ·

Sep. 25, 21 · Tutorial

Like (5)

Save

14.8K Views

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
   
 

   <?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.

    Java
   
   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.

    Java
   
   @Bean
public MessageChannel pubsubInputChannel() {
    return new DirectChannel();
}

Then add the inbound channel adapter. The ack mode will be set to manual.

    Java
   
 

   @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.

    Java
   
 

   @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:

    Java
   
 

   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.

    Java
   
 

   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

    Java
   
   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

    Shell
   
   spring.cloud.gcp.pubsub.enabled=true

This 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.

application Spring Framework

Published at DZone with permission of Emmanouil Gkatziouras, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

Trending

Receive Pub/Sub Messages to Your Spring Application

Time to dive into GCP Pub/Sub.

Related

Partner Resources