Subscribing to Google PubSub Topics with Spring Boot GCP

This is the continuation of my previous article Retrieving conversation history from Dialogflow. There, we exported Dialogflow logs to Google Stackdriver and then routed them to a Google PubSub Topic. Also we setup Google Cloud SDK in our local machine and subscribed to the topic with gcloud commands. In this article we’ll implement a Spring Boot application which can subscribe to our PubSub topic.

Starting with Spring Boot

Since this is not a Spring Boot tutorial, I will not be explaining how you can set up your Spring Boot application. See https://spring.io/guides/gs/spring-boot/ to get started. The option I prefer to set up a Spring Boot application is to use the Spring Initializer.

This will be my pom.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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <parent>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-parent</artifactId>
      <version>2.2.6.RELEASE</version>
      <relativePath/> <!-- lookup parent from repository -->
   </parent>
   <groupId>com.dialog</groupId>
   <artifactId>chatboard</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   <name>chatboard</name>
   <description>PubSub subscriber and Dashboard backend</description>
   <properties>
      <java.version>1.8</java.version>
      <spring-cloud.version>Hoxton.SR3</spring-cloud.version>
      <elasticsearch.version>7.3.0</elasticsearch.version>
   </properties>
   <dependencies>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-web</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-gcp-starter</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.integration</groupId>
         <artifactId>spring-integration-core</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-test</artifactId>
         <scope>test</scope>
         <exclusions>
            <exclusion>
               <groupId>org.junit.vintage</groupId>
               <artifactId>junit-vintage-engine</artifactId>
            </exclusion>
         </exclusions>
      </dependency>
   </dependencies>
   <dependencyManagement>
      <dependencies>
         <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
         </dependency>
      </dependencies>
   </dependencyManagement>
   <build>
      <plugins>
         <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
         </plugin>
      </plugins>
   </build>
</project>

Spring Cloud GCP

Spring Cloud GCP offers a wide collection of libraries that make it easier to use Google Cloud Platform from Spring Framework applications. We will be using the Spring Cloud GCP Pub/Sub Support (Spring Integration and Spring Cloud Stream Binder). Check this documentation for detailed information.

Google Cloud Service Account Credentials

Remember our Stackdriver logs and the PubSub topic from the previous article? All these reside in a Google Cloud Project and my project was CHITTI. Each project can have several service accounts and we’ll need the credentials of the relavant Service Account as a json file, in order to communicate to the project from our Spring Boot application.

From the side menu in Google Cloud console, select IAM & Admin -> Service Accounts .

Click on ⋮ next to the service account with the name “App Engine default service account”, then select Create key.

In the next window, select JSON as the Key type and click CREATE. A Json file will be downloaded to your computer. Copy the file to a folder you prefer.

Subscribing to Google PubSub Topic

In your Spring Boot application, create an application.properties file in the resources folder, then add the following. Replace project-Id with your Google Cloud project id and replace path-to-json-file with path to the Json file you just downloaded.

spring.cloud.gcp.project-id=project-Id
spring.cloud.gcp.credentials.location=file:path-to-json-file

Create a Config file and add the following code. Replace subscription-name with the name you gave to your subscription in our previous article.

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

@Bean
public PubSubInboundChannelAdapter messageChannelAdapter(
        @Qualifier("pubsubInputChannel") MessageChannel inputChannel,
        PubSubTemplate pubSubTemplate) {
    PubSubInboundChannelAdapter adapter =
            new PubSubInboundChannelAdapter(pubSubTemplate, "subscription-name");
    adapter.setOutputChannel(inputChannel);

    return adapter;
}

In the main class, add the below. Here DemoApplication is my main class. This will log the messages received from PubSub to your console. Actually, these messages are the Conversation History from your Dialogflow chatbot.

private static final Log LOGGER = LogFactory.getLog(DemoApplication.class);

@ServiceActivator(inputChannel = "pubsubInputChannel")
public void messageReceiver(String payload) {
    LOGGER.info("Message arrived! Payload: " + payload);
}

We didn’t do hours of work just to log the conversation history in the console, aren’t we? We can send this incoming data to a service and then do whatever we want with the data. In my case, I wanted to write these messages to Elasticsearch.

private static final Log LOGGER = LogFactory.getLog(DemoApplication.class);

@ServiceActivator(inputChannel = "pubsubInputChannel")
public void messageReceiver(String payload) {
    pubsubService.saveToElasticsearch(payload);
}