{"id":85154,"date":"2018-12-28T10:00:19","date_gmt":"2018-12-28T08:00:19","guid":{"rendered":"https:\/\/www.javacodegeeks.com\/?p=85154"},"modified":"2019-01-11T14:13:17","modified_gmt":"2019-01-11T12:13:17","slug":"bootiful-spring-cloud-stream","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2018\/12\/bootiful-spring-cloud-stream.html","title":{"rendered":"Bootiful GCP: Spring Cloud Stream with Google Cloud Pub\/Sub"},"content":{"rendered":"

I\u2019ve recently read Josh Long<\/a>\u2018s Bootiful GCP series<\/a> on Sprint Central\u2019s engineering blog and especially liked the 4th part<\/a> about using Google Cloud\u2019s Pub\/Sub. I felt inspired by the series and as I\u2019m also evaluating Spring Cloud Stream for a new project of mine. I thought, I would expand on that article where Josh left off. This article describes how to use Spring Cloud Stream<\/a> with Google Cloud Pub\/Sub for implementing a simple producer and a consumer application.<\/p>\n

Introduction<\/h2>\n

You can safely skip this part if you\u2019ve read Josh\u2019s article before. If you haven\u2019t done so, no worries, I\u2019ll quickly summarize some key points here.<\/p>\n

What is Google Cloud Pub\/Sub?<\/h3>\n

Google defines Pub\/Sub<\/a> in the following way.<\/p>\n

\n

Cloud Pub\/Sub brings the scalability, flexibility, and reliability of enterprise message-oriented middleware to the cloud. By providing many-to-many, asynchronous messaging that decouples senders and receivers, it allows for secure and highly available communication between independently written applications.<\/p>\n<\/blockquote>\n

\n
\nhttps:\/\/cloud.google.com\/pubsub\/docs\/overview\n<\/div>\n<\/figure>\n

Simply put, Pub\/Sub is Google\u2019s solution for supporting developers connecting application components with a message broker at Google\u2019s scale. As the name suggests this solution implements publish\/subscribe mechanism with the same concepts you would expect. Messages can be submitted to topics and all the subscribers of a certain topic receive a published message.<\/p>\n

\n
\"Google<\/figure>\n<\/div>\n

It\u2019s important to emphasize here that Pub\/Sub offers at least once delivery for each submitted message. If you want to ensure that a message gets delivered only once, then you would have to take care of that yourself. <\/strong><\/p>\n

What is Spring Integration?<\/h3>\n

Spring Integration<\/a> is a Spring project in their portfolio. An entire article or even an entire book could be written on it, as it\u2019s a vast framework in itself. In summary, Spring Integration is a framework which help you design and integrate applications by using EIP patterns<\/a>. The two most basic primitives Spring Integration<\/em> is built upon are Message<T><\/code> and MessageChannel<\/code>. In this regard, developers can decouple and isolate components from each other. You can think of this mechanism as though Spring Integration would take the idea of dependency injection even further in a way where components don\u2019t even have to know about each other, but they\u2019re exchanging messages instead.<\/p>\n

Channels can connect components with each other either if they live in the same JVM or even if they\u2019re distributed and separated by the network. At this point the relevant concept to understand is what channel adapters are. They\u2019re basically meant to transform a Spring Framework message as it goes through a message channel, into a piece of data that can be used by external systems.<\/p>\n

\n
\"Google<\/figure>\n<\/div>\n

A myriad of adapters are provided by Spring Integration which help developers connect to databases, message brokers, and to many other external systems. In this case, adapters are being used for submitting and receiving messages to\/from Google Cloud Pub\/Sub. The Spring Cloud GCP<\/a> project provides in- and outbound adapters for Pub\/Sub and that makes message exchanges transparent from the point of view of a Spring Integration message flow.<\/p>\n

If you read Josh\u2019s article<\/a> what he does is that he\u2019s introducing Spring Integration for using Pub\/Sub in a clean and consistent way. That means that direct references of PubSubTemplate<\/a> are removed, as a consequence if you wanted to adapt examples in that article for example to RabbitMQ, all you would have to do is just replace the channel adapters accordingly.

<\/div> <\/div><\/p>\n

What is Spring Cloud Stream?<\/h3>\n

Messaging is a really great fit for the microservices world where a set of distributed components communicate with each other. As messages and channels are first class citizens in Spring Integration, it\u2019s a great fit for that. On the other hand, Spring Integration was specifically designed to implement those EIP patterns.<\/p>\n

However, with modern application development, we don\u2019t necessarily want to integrate with legacy systems, we\u2019d rather integrate with modern message brokers like RabbitMQ<\/a>, Apache Kafka<\/a> or with GCP Pub\/Sub<\/a> in this case. That said, we don\u2019t need the full repertoire of Spring Integration in terms of being able to integrate with a wide variety of external systems. That extra flexibility would require us to configure adapters, which we don\u2019t need. If we\u2019re just using GCP Pub\/Sub or any other modern message broker previously mentioned, it becomes tedious having to define and configure the adapters for every single component.<\/p>\n

\n
\"Google<\/figure>\n<\/div>\n

We do want the flexibility of being able to work with messages and we want to take advantage of using a message broker, but we don\u2019t want to write that much code bare Spring Integration would require. Spring Cloud Stream<\/a> builds on Spring Integration and it leverages the same primitives like messages and channels, but it off-loads the developer from having to wire these components together; as channels are connected to external brokers through middleware-specific Binder<\/em> implementations.<\/p>\n

Using Spring Cloud Stream with Google Cloud Pub\/Sub<\/h2>\n

I think I\u2019ve talked enough about the background of Spring Cloud Stream, Spring Integration and Google Cloud Pub\/Sub. It\u2019s time to see some code. There are two very simple Spring Boot applications, which exchange a simple string as the payload of messages. Let\u2019s start with the publisher<\/a>.<\/p>\n

Publisher<\/h3>\n

This is basically a simple controller which send a simple String as the message\u2019s payload. If you\u2019ve worked with Spring Integration before, there\u2019s nothing special about the sending part.<\/p>\n

@RestController\npublic class PublisherController {\n\n  private final MessageChannel outgoing;\n\n  public PublisherController(Channels channels) {\n    outgoing = channels.outgoing();\n  }\n\n  @PostMapping(\"\/publish\/{name}\")\n  public void publish(@PathVariable String name) {\n    outgoing.send(MessageBuilder.withPayload(\"Hello \" + name + \"!\").build());\n  }\n\n}<\/pre>\n
What\u2019s interesting is how message channels are bound to the resources of an actual message broker. In line 6-8 a bean (Channels<\/code>) is injected and that seems to hold a reference to the outgoing message channel.<\/p>\n
import org.springframework.cloud.stream.annotation.Output;\nimport org.springframework.messaging.MessageChannel;\n\npublic interface Channels {\n\n  @Output\n  MessageChannel outgoing();\n\n}<\/pre>\n
Channels<\/code> in turn is just an interface where arbitrary number of message channels can be defined and marked with either @Input<\/code> or @Output<\/code>. Spring Cloud Stream takes care of instantiating a proxy object which is responsible for returning references to MessageChannel<\/code> objects.<\/p>\n
@EnableBinding(Channels.class)\n@SpringBootApplication\npublic class PubsubPublisherApplication {\n\n  public static void main(String[] args) {\n    SpringApplication.run(PubsubPublisherApplication.class, args);\n  }\n\n}<\/pre>\n
Spring Cloud Stream relies on both Spring Boot and Spring Integration. The @EnableBinding<\/code> annotation marks Channels<\/code> as a bindable interface and pairs a logical binding name (outgoing<\/code>) with a destination. What destination means that varies across binders, for Pub\/Sub it means a topic for a message producer and a subscription for a message consumer. These bindings can be defined in application.yml<\/code>.<\/p>\n
spring:\n  cloud:\n    stream:\n      bindings:\n        outgoing:\n          destination: reservations<\/pre>\n
Subscriber<\/h3>\n
The subscriber<\/a> is even simpler than the publisher, it\u2019s just a single class.<\/p>\n
import org.springframework.boot.autoconfigure.SpringBootApplication;\nimport org.springframework.cloud.stream.annotation.EnableBinding;\nimport org.springframework.cloud.stream.annotation.StreamListener;\nimport org.springframework.cloud.stream.messaging.Sink;\nimport org.springframework.messaging.Message;\n\n@Slf4j\n@EnableBinding(Sink.class)\n@SpringBootApplication\npublic class PubsubSubscriberApplication {\n\n  public static void main(String[] args) {\n    SpringApplication.run(PubsubSubscriberApplication.class, args);\n  }\n\n  @StreamListener(Sink.INPUT)\n  public void handleMessage(Message<String> message) {\n    log.info(\"Received: {}.\", message.getPayload());\n  }\n\n}<\/pre>\n
What\u2019s worth mentioning here is what Sink is? As we\u2019ve just seen @EnableBinding<\/code> can take interfaces and then the framework hides the complexity of wiring in- and outbound message adapters to message channels and it also configures the related infrastructure. Most applications just send or receive messages to\/from a single channel. That\u2019s why Spring Cloud Stream provides the Source<\/code>, Sink<\/code>, and Processor<\/code> interfaces in order to help you cut down on code. That said, we could\u2019ve also used a Source<\/code> for the publisher instead of defining Channels<\/code>, but I wanted to show what the framework is capable of.<\/p>\n
Running the demo<\/h2>\n
In order to be able to run the examples, you\u2019ll need the complete the following steps.<\/p>\n
    \n
  1. \n
      \n
    1. \n
      Create a GCP project<\/a><\/h2>\n
      If you have one already, you can skip this step.<\/p>\n<\/li>\n
    2. \n
      Start Google Cloud Shell<\/a><\/h2>\n
      I think easier if you don\u2019t have to install anything. Google Cloud Shell<\/a> comes with Google Cloud SDK<\/a>, Git, Maven and Java pre-installed by default.<\/p>\n<\/li>\n
    3. \n
      Enable Pub\/Sub API<\/h2>\n
      As Spring Cloud Stream is an opinionated framework, applications built upon it will create topics and subscriptions on their own. That said, creating a topic and subscription manually is optional here. You\u2019ll have to enable the Pub\/Sub API though.<\/p>\n
      % gcloud services enable pubsub.googleapis.com\n% gcloud pubsub topics create reservations\n% gcloud pubsub subscriptions create reservations --topic=reservations<\/pre>\n<\/li>\n
    4. \n
      Clone the springuni-examples<\/a> repo<\/h2>\n
      % git clone https:\/\/github.com\/springuni\/springuni-examples.git<\/pre>\n<\/li>\n
    5. \n
      Start the publisher<\/h2>\n
      % cd ~\/springuni-examples\/spring-cloud\/spring-cloud-stream-pubsub-publisher\n% mvn spring-boot:run<\/pre>\n<\/li>\n
    6. \n
      Start the subscriber<\/h2>\n
      Google Cloud Shell comes with tmux support and that also means that it starts a tmux session by default. That can be disabled of course. Important point is that you don\u2019t have to open a new shell, you just have to open a new window by hitting Ctrl-B and C. Refer to Tmux Key Bindings<\/a> for further details.<\/p>\n
      % cd ~\/springuni-examples\/spring-cloud\/spring-cloud-stream-pubsub-subscriber\n% mvn spring-boot:run<\/pre>\n<\/li>\n
    7. \n
      Send a message<\/h2>\n
      Open a new window again as before and send a message.<\/p>\n
      % curl -XPOST http:\/\/localhost:8080\/publish\/test<\/pre>\n
      You should see the subscriber receiving it.<\/p>\n<\/li>\n<\/ol>\n<\/li>\n<\/ol>\n
      Questions<\/h2>\n