Messaging with Spring Integration and Google Cloud Pub/Sub

1. Overview

Spring Integration provides you with a messaging mechanism to exchange Messages through MessageChannels. It uses channel adapters to communicate with external systems.

In this exercise, we will create two apps that communicate using the Spring Integration channel adapters provided by Spring Cloud GCP. These adapters make Spring Integration use Google Cloud Pub/Sub as the message exchange backend.

You will learn how to use Cloud Shell and the Cloud SDK gcloud command.

This tutorial uses the sample code from the Spring Boot Getting Started guide.

What you'll learn

• How to exchange messages between apps with Google Cloud Pub/Sub using Spring Integration and Spring Cloud GCP

What you'll need

• A Google Cloud Platform Project
• A Browser, such Chrome or Firefox
• Familiarity with standard Linux text editors such as Vim, EMACs or Nano

How will you use use this tutorial?

Read it through only Read it and complete the exercises

How would you rate your experience with building HTML/CSS web apps?

Novice Intermediate Proficient

How would you rate your experience with using Google Cloud Platform services?

Novice Intermediate Proficient

2. Setup and Requirements

Self-paced environment setup

1. Sign-in to the Google Cloud Console and create a new project or reuse an existing one. If you don't already have a Gmail or Google Workspace account, you must create one.

• The Project name is the display name for this project's participants. It is a character string not used by Google APIs. You can always update it.
• The Project ID is unique across all Google Cloud projects and is immutable (cannot be changed after it has been set). The Cloud Console auto-generates a unique string; usually you don't care what it is. In most codelabs, you'll need to reference your Project ID (typically identified as PROJECT_ID). If you don't like the generated ID, you might generate another random one. Alternatively, you can try your own, and see if it's available. It can't be changed after this step and remains for the duration of the project.
• For your information, there is a third value, a Project Number, which some APIs use. Learn more about all three of these values in the documentation.

Caution: A project ID is globally unique and can't be used by anyone else after you've selected it. You are the only user of that ID. Even if a project is deleted, the ID can't be used again

Note: If you use a Gmail account, you can leave the default location set to No organization. If you use a Google Workspace account, choose a location that makes sense for your organization.

2. Next, you'll need to enable billing in the Cloud Console to use Cloud resources/APIs. Running through this codelab won't cost much, if anything at all. To shut down resources to avoid incurring billing beyond this tutorial, you can delete the resources you created or delete the project. New Google Cloud users are eligible for the $300 USD Free Trial program.

Google Cloud Shell

While Google Cloud can be operated remotely from your laptop, in this codelab we will be using Google Cloud Shell, a command line environment running in the Cloud.

Activate Cloud Shell

1. From the Cloud Console, click Activate Cloud Shell .

If this is your first time starting Cloud Shell, you're presented with an intermediate screen describing what it is. If you were presented with an intermediate screen, click Continue.

It should only take a few moments to provision and connect to Cloud Shell.

This virtual machine is loaded with all the development tools needed. It offers a persistent 5 GB home directory and runs in Google Cloud, greatly enhancing network performance and authentication. Much, if not all, of your work in this codelab can be done with a browser.

Once connected to Cloud Shell, you should see that you are authenticated and that the project is set to your project ID.

2. Run the following command in Cloud Shell to confirm that you are authenticated:

gcloud auth list

Command output

 Credentialed Accounts
ACTIVE  ACCOUNT
*       <my_account>@<my_domain.com>

To set the active account, run:
    $ gcloud config set account `ACCOUNT`

Note: The gcloud command-line tool is the powerful and unified command-line tool in Google Cloud. It comes preinstalled in Cloud Shell. You will notice its support for tab completion. For more information, see gcloud command-line tool overview.

3. Run the following command in Cloud Shell to confirm that the gcloud command knows about your project:

gcloud config list project

Command output

[core]
project = <PROJECT_ID>

If it is not, you can set it with this command:

gcloud config set project <PROJECT_ID>

Command output

Updated property [core/project].

Note: If you're completing this tutorial outside of Cloud Shell, follow Set up Application Default Credentials.

3. Provision Pub/Sub resources

Navigate to the Google Cloud Pub/Sub topics page.

Click Create Topic.

Type exampleTopic as the name of the topic and then click Create..

After the topic is created, remain in the Topics page. Look for the topic you just created, press the three vertical dots at the end of the line and click New Subscription.

Type exampleSubscription in the subscription name text box and click Create.

Alternatively, you can also use the gcloud CLI to create a new topic and subscription, for example:

$ gcloud pubsub topics create exampleTopic

$ gcloud pubsub subscriptions create exampleSubscription --topic exampleTopic

4. Initialize Spring Boot Applications

After Cloud Shell launches, you can use the command line to generate two new Spring Boot applications with Spring Initializr:

$ curl https://start.spring.io/starter.tgz \
  -d bootVersion=3.0.5 \
  -d dependencies=web,integration,cloud-gcp-pubsub \
  -d type=maven-project \
  -d baseDir=spring-integration-sender | tar -xzvf -

$ curl https://start.spring.io/starter.tgz \
  -d bootVersion=3.0.5 \
  -d dependencies=web,integration,cloud-gcp-pubsub \
  -d type=maven-project \
  -d baseDir=spring-integration-receiver | tar -xzvf -

5. Create an Application to Send Messages

Now let's create our message sending app. Change to the directory of the sending app.

$ cd spring-integration-sender

We want our app to write messages to a channel. After a message is in the channel, it will get picked up by the outbound channel adapter, which converts it from a generic Spring message to a Google Cloud Pub/Sub message and publishes it to a Google Cloud Pub/Sub topic.

In order for our app to write to a channel, we can use a Spring Integration messaging gateway. Using a text editor from vim, emacs or nano, declare a PubsubOutboundGateway interface inside the DemoApplication class.

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.integration.annotation.MessagingGateway;

@SpringBootApplication
public class DemoApplication {

  ...

  @MessagingGateway(defaultRequestChannel = "pubsubOutputChannel")
  public interface PubsubOutboundGateway {
    void sendToPubsub(String text);
  }
}

We now have a mechanism to send messages to a channel, but where do those messages go after they are in the channel?

We need an outbound channel adapter to consume new messages in the channel and publish them to a Google Cloud Pub/Sub topic.

src/main/java/com/example/demo/DemoApplication.java

...
import com.google.cloud.spring.pubsub.core.PubSubTemplate;
import com.google.cloud.spring.pubsub.integration.outbound.PubSubMessageHandler;

import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.MessageHandler;

@SpringBootApplication
public class DemoApplication {

  ...

  @Bean
  @ServiceActivator(inputChannel = "pubsubOutputChannel")
  public MessageHandler messageSender(PubSubTemplate pubsubTemplate) {
    return new PubSubMessageHandler(pubsubTemplate, "exampleTopic");
  }
}

The @ServiceActivator annotation causes this MessageHandler to be applied to any new messages in inputChannel. In this case, we are calling our outbound channel adapter, PubSubMessageHandler, to publish the message to Google Cloud Pub/Sub's exampleTopic topic.

With the channel adapter in place, we can now auto-wire a PubsubOutboundGateway object and use it to write a message to a channel.

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.servlet.view.RedirectView;

@SpringBootApplication
public class DemoApplication {

  ...

  @Autowired
  private PubsubOutboundGateway messagingGateway;

  @PostMapping("/postMessage")
  public RedirectView postMessage(@RequestParam("message") String message) {
    this.messagingGateway.sendToPubsub(message);
    return new RedirectView("/");
  }
}

Because of the @PostMapping annotation, we now have an endpoint that is listening to HTTP POST requests, but not without also adding a @RestController annotation to the DemoApplication class to mark it as a REST controller.

src/main/java/com/example/demo/DemoApplication.java

import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
@RestController
public class DemoApplication {
  ...
}

Make sure JAVA_HOME is set to the right version.

export JAVA_HOME=/usr/lib/jvm/java-1.17.0-openjdk-amd64

Run the sender app.

# Set the Project ID in environmental variable
$ export GOOGLE_CLOUD_PROJECT=`gcloud config list --format 'value(core.project)'`

$ ./mvnw spring-boot:run

The app is listening to POST requests containing a message on port 8080 and endpoint /postMessage, but we will get to this later.

6. Create an Application to Receive Messages

We just created an app that sends messages through Google Cloud Pub/Sub. Now, we will create another app that receives those messages and processes them.

Click + to open a new Cloud Shell session.

Then, in the new Cloud Shell session, change directories to the receiver app's directory:

$ cd spring-integration-receiver

In the previous app, the messaging gateway declaration created the outbound channel for us. Since we don't use a messaging gateway to receive messages, we need to declare our own MessageChannel where incoming messages will arrive.

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;

@SpringBootApplication
public class DemoApplication {

  ...

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

We will need the inbound channel adapter to receive messages from Google Cloud Pub/Sub and relay them to pubsubInputChannel.

src/main/java/com/example/demo/DemoApplication.java

...
import com.google.cloud.spring.pubsub.core.PubSubTemplate;
import com.google.cloud.spring.pubsub.integration.inbound.PubSubInboundChannelAdapter;

import org.springframework.beans.factory.annotation.Qualifier;

@SpringBootApplication
public class DemoApplication {

  ...

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

    return adapter;
  }
}

This adapter binds itself to the pubsubInputChannel and listens to new messages from the Google Cloud Pub/Sub exampleSubscription subscription.

We have a channel where incoming messages are posted to, but what to do with those messages?

Let's process them with a @ServiceActivator that is triggered when new messages arrive at pubsubInputChannel. In this case, we'll just log the message payload.

src/main/java/com/example/demo/DemoApplication.java

...
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.annotation.ServiceActivator;

@SpringBootApplication
public class DemoApplication {

  ...

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

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

Make sure JAVA_HOME is set to the right version.

export JAVA_HOME=/usr/lib/jvm/java-1.17.0-openjdk-amd64

Run the receiver app.

$ ./mvnw spring-boot:run -Dspring-boot.run.jvmArguments="-Dserver.port=8081"

Now any messages you send to the sender app will be logged on the receiver app. To test that, open a new Cloud Shell session and make a HTTP POST request to the sender app.

$ curl --data "message=Hello world!" localhost:8080/postMessage

Then, verify that the receiver app logged the message you sent!

INFO: Message arrived! Payload: Hello world!

7. Cleanup

Delete the subscription and topic created as part of this exercise.

$ gcloud pubsub subscriptions delete exampleSubscription
$ gcloud pubsub topics delete exampleTopic

8. Summary

You set up two Spring Boot apps that use the Spring Integration Channel Adapters for Google Cloud Pub/Sub. They exchange messages among themselves without ever interacting with the Google Cloud Pub/Sub API.

9. Congratulations!

You learned how to use the Spring Integration Channel Adapters for Google Cloud Pub/Sub!

Learn More

• Google Cloud Pub/Sub: https://cloud.google.com/pubsub/
• Spring on GCP project: http://cloud.spring.io/spring-cloud-gcp/
• Spring on GCP GitHub repository: https://github.com/GoogleCloudPlatform/spring-cloud-gcp
• Java on Google Cloud Platform: https://cloud.google.com/java/

License

This work is licensed under a Creative Commons Attribution 2.0 Generic License.