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 be aware that 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>

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

 This post is ad-supported
​