Spring Integration

Spring Cloud GCP provides Spring Integration adapters that allow your applications to use Enterprise Integration Patterns backed up by Google Cloud Platform services.

Channel Adapters for Cloud Pub/Sub

The channel adapters for Google Cloud Pub/Sub connect your Spring MessageChannels to Google Cloud Pub/Sub topics and subscriptions. This enables messaging between different processes, applications or micro-services backed up by Google Cloud Pub/Sub.

The Spring Integration Channel Adapters for Google Cloud Pub/Sub are included in the spring-cloud-gcp-pubsub module.

Maven coordinates, using Spring Cloud GCP BOM:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-gcp-pubsub</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-core</artifactId>
</dependency>

Gradle coordinates:

dependencies {
    compile group: 'org.springframework.cloud', name: 'spring-cloud-gcp-pubsub'
    compile group: 'org.springframework.integration', name: 'spring-integration-core'
}

Inbound channel adapter (using Pub/Sub Streaming Pull)

PubSubInboundChannelAdapter is the inbound channel adapter for GCP Pub/Sub that listens to a GCP Pub/Sub subscription for new messages. It converts new messages to an internal Spring Message and then sends it to the bound output channel.

Google Pub/Sub treats message payloads as byte arrays. So, by default, the inbound channel adapter will construct the Spring Message with byte[] as the payload. However, you can change the desired payload type by setting the payloadType property of the PubSubInboundChannelAdapter. The PubSubInboundChannelAdapter delegates the conversion to the desired payload type to the PubSubMessageConverter configured in the PubSubTemplate.

To use the inbound channel adapter, a PubSubInboundChannelAdapter must be provided and configured on the user application side.

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

@Bean
public PubSubInboundChannelAdapter messageChannelAdapter(
    @Qualifier("pubsubInputChannel") MessageChannel inputChannel,
    SubscriberFactory subscriberFactory) {
    PubSubInboundChannelAdapter adapter =
        new PubSubInboundChannelAdapter(subscriberFactory, "subscriptionName");
    adapter.setOutputChannel(inputChannel);
    adapter.setAckMode(AckMode.MANUAL);

    return adapter;
}

In the example, we first specify the MessageChannel where the adapter is going to write incoming messages to. The MessageChannel implementation isn’t important here. Depending on your use case, you might want to use a MessageChannel other than PublishSubscribeChannel.

Then, we declare a PubSubInboundChannelAdapter bean. It requires the channel we just created and a SubscriberFactory, which creates Subscriber objects from the Google Cloud Java Client for Pub/Sub. The Spring Boot starter for GCP Pub/Sub provides a configured SubscriberFactory.

The PubSubInboundChannelAdapter supports three acknowledgement modes, with AckMode.AUTO being the default value;

Automatic acking (AckMode.AUTO)

A message is acked with GCP Pub/Sub if the adapter sent it to the channel and no exceptions were thrown. If a RuntimeException is thrown while the message is processed, then the message is nacked.

Automatic acking OK (AckMode.AUTO_ACK)

A message is acked with GCP Pub/Sub if the adapter sent it to the channel and no exceptions were thrown. If a RuntimeException is thrown while the message is processed, then the message is neither acked / nor nacked.

This is useful when using the subscription’s ack deadline timeout as a retry delivery backoff mechanism.

Manually acking (AckMode.MANUAL)

The adapter attaches a BasicAcknowledgeablePubsubMessage object to the Message headers. Users can extract the BasicAcknowledgeablePubsubMessage using the GcpPubSubHeaders.ORIGINAL_MESSAGE key and use it to (n)ack a message.

@Bean
@ServiceActivator(inputChannel = "pubsubInputChannel")
public MessageHandler messageReceiver() {
    return message -> {
        LOGGER.info("Message arrived! Payload: " + new String((byte[]) message.getPayload()));
        BasicAcknowledgeablePubsubMessage originalMessage =
              message.getHeaders().get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage.class);
        originalMessage.ack();
    };
}

Pollable Message Source (using Pub/Sub Synchronous Pull)

While PubSubInboundChannelAdapter, through the underlying Asynchronous Pull Pub/Sub mechanism, provides the best performance for high-volume applications that receive a steady flow of messages, it can create load balancing anomalies due to message caching. This behavior is most obvious when publishing a large batch of small messages that take a long time to process individually. It manifests as one subscriber taking up most messages, even if multiple subscribers are available to take on the work. For a more detailed explanation of this scenario, see GCP Pub/Sub documentation.

In such a scenario, a PubSubMessageSource can help spread the load between different subscribers more evenly.

As with the Inbound Channel Adapter, the message source has a configurable acknowledgement mode, payload type, and header mapping.

The default behavior is to return from the synchronous pull operation immediately if no messages are present. This can be overridden by using setBlockOnPull() method to wait for at least one message to arrive.

By default, PubSubMessageSource pulls from the subscription one message at a time. To pull a batch of messages on each request, use the setMaxFetchSize() method to set the batch size.

@Bean
@InboundChannelAdapter(channel = "pubsubInputChannel", poller = @Poller(fixedDelay = "100"))
public MessageSource<Object> pubsubAdapter(PubSubTemplate pubSubTemplate) {
	PubSubMessageSource messageSource = new PubSubMessageSource(pubSubTemplate,  "exampleSubscription");
	messageSource.setAckMode(AckMode.MANUAL);
	messageSource.setPayloadType(String.class);
	messageSource.setBlockOnPull(true);
	messageSource.setMaxFetchSize(100);
	return messageSource;
}

The @InboundChannelAdapter annotation above ensures that the configured MessageSource is polled for messages, which are then available for manipulation with any Spring Integration mechanism on the pubsubInputChannel message channel. For example, messages can be retrieved in a method annotated with @ServiceActivator, as seen below.

For additional flexibility, PubSubMessageSource attaches an AcknowledgeablePubSubMessage object to the GcpPubSubHeaders.ORIGINAL_MESSAGE message header. The object can be used for manually (n)acking the message.

@ServiceActivator(inputChannel = "pubsubInputChannel")
public void messageReceiver(String payload,
        @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) AcknowledgeablePubsubMessage message)
            throws InterruptedException {
    LOGGER.info("Message arrived by Synchronous Pull! Payload: " + payload);
    message.ack();
}

AcknowledgeablePubSubMessage objects acquired by synchronous pull are aware of their own acknowledgement IDs. Streaming pull does not expose this information due to limitations of the underlying API, and returns BasicAcknowledgeablePubsubMessage objects that allow acking/nacking individual messages, but not extracting acknowledgement IDs for future processing.

Outbound channel adapter

PubSubMessageHandler is the outbound channel adapter for GCP Pub/Sub that listens for new messages on a Spring MessageChannel. It uses PubSubTemplate to post them to a GCP Pub/Sub topic.

To construct a Pub/Sub representation of the message, the outbound channel adapter needs to convert the Spring Message payload to a byte array representation expected by Pub/Sub. It delegates this conversion to the PubSubTemplate. To customize the conversion, you can specify a PubSubMessageConverter in the PubSubTemplate that should convert the Object payload and headers of the Spring Message to a PubsubMessage.

To use the outbound channel adapter, a PubSubMessageHandler bean must be provided and configured on the user application side.

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

The provided PubSubTemplate contains all the necessary configuration to publish messages to a GCP Pub/Sub topic.

PubSubMessageHandler publishes messages asynchronously by default. A publish timeout can be configured for synchronous publishing. If none is provided, the adapter waits indefinitely for a response.

It is possible to set user-defined callbacks for the publish() call in PubSubMessageHandler through the setPublishFutureCallback() method. These are useful to process the message ID, in case of success, or the error if any was thrown.

To override the default destination you can use the GcpPubSubHeaders.DESTINATION header.

@Autowired
private MessageChannel pubsubOutputChannel;

public void handleMessage(Message<?> msg) throws MessagingException {
    final Message<?> message = MessageBuilder
        .withPayload(msg.getPayload())
        .setHeader(GcpPubSubHeaders.TOPIC, "customTopic").build();
    pubsubOutputChannel.send(message);
}

It is also possible to set an SpEL expression for the topic with the setTopicExpression() or setTopicExpressionString() methods.

Header mapping

These channel adapters contain header mappers that allow you to map, or filter out, headers from Spring to Google Cloud Pub/Sub messages, and vice-versa. By default, the inbound channel adapter maps every header on the Google Cloud Pub/Sub messages to the Spring messages produced by the adapter. The outbound channel adapter maps every header from Spring messages into Google Cloud Pub/Sub ones, except the ones added by Spring, like headers with key "id", "timestamp" and "gcp_pubsub_acknowledgement". In the process, the outbound mapper also converts the value of the headers into string.

Each adapter declares a setHeaderMapper() method to let you further customize which headers you want to map from Spring to Google Cloud Pub/Sub, and vice-versa.

For example, to filter out headers "foo", "bar" and all headers starting with the prefix "prefix_", you can use setHeaderMapper() along with the PubSubHeaderMapper implementation provided by this module.

PubSubMessageHandler adapter = ...
...
PubSubHeaderMapper headerMapper = new PubSubHeaderMapper();
headerMapper.setOutboundHeaderPatterns("!foo", "!bar", "!prefix_*", "*");
adapter.setHeaderMapper(headerMapper);

The order in which the patterns are declared in PubSubHeaderMapper.setOutboundHeaderPatterns() and PubSubHeaderMapper.setInboundHeaderPatterns() matters. The first patterns have precedence over the following ones.

In the previous example, the "*" pattern means every header is mapped. However, because it comes last in the list, the previous patterns take precedence.

Sample

Available examples:

Channel Adapters for Google Cloud Storage

The channel adapters for Google Cloud Storage allow you to read and write files to Google Cloud Storage through MessageChannels.

Spring Cloud GCP provides two inbound adapters, GcsInboundFileSynchronizingMessageSource and GcsStreamingMessageSource, and one outbound adapter, GcsMessageHandler.

The Spring Integration Channel Adapters for Google Cloud Storage are included in the spring-cloud-gcp-storage module.

To use the Storage portion of Spring Integration for Spring Cloud GCP, you must also provide the spring-integration-file dependency, since it isn’t pulled transitively.

Maven coordinates, using Spring Cloud GCP BOM:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-gcp-storage</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-file</artifactId>
</dependency>

Gradle coordinates:

dependencies {
    compile group: 'org.springframework.cloud', name: 'spring-cloud-gcp-starter-storage'
    compile group: 'org.springframework.integration', name: 'spring-integration-file'
}

Inbound channel adapter

The Google Cloud Storage inbound channel adapter polls a Google Cloud Storage bucket for new files and sends each of them in a Message payload to the MessageChannel specified in the @InboundChannelAdapter annotation. The files are temporarily stored in a folder in the local file system.

Here is an example of how to configure a Google Cloud Storage inbound channel adapter.

@Bean
@InboundChannelAdapter(channel = "new-file-channel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> synchronizerAdapter(Storage gcs) {
  GcsInboundFileSynchronizer synchronizer = new GcsInboundFileSynchronizer(gcs);
  synchronizer.setRemoteDirectory("your-gcs-bucket");

  GcsInboundFileSynchronizingMessageSource synchAdapter =
          new GcsInboundFileSynchronizingMessageSource(synchronizer);
  synchAdapter.setLocalDirectory(new File("local-directory"));

  return synchAdapter;
}

Inbound streaming channel adapter

The inbound streaming channel adapter is similar to the normal inbound channel adapter, except it does not require files to be stored in the file system.

Here is an example of how to configure a Google Cloud Storage inbound streaming channel adapter.

@Bean
@InboundChannelAdapter(channel = "streaming-channel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<InputStream> streamingAdapter(Storage gcs) {
  GcsStreamingMessageSource adapter =
          new GcsStreamingMessageSource(new GcsRemoteFileTemplate(new GcsSessionFactory(gcs)));
  adapter.setRemoteDirectory("your-gcs-bucket");
  return adapter;
}

Outbound channel adapter

The outbound channel adapter allows files to be written to Google Cloud Storage. When it receives a Message containing a payload of type File, it writes that file to the Google Cloud Storage bucket specified in the adapter.

Here is an example of how to configure a Google Cloud Storage outbound channel adapter.

@Bean
@ServiceActivator(inputChannel = "writeFiles")
public MessageHandler outboundChannelAdapter(Storage gcs) {
  GcsMessageHandler outboundChannelAdapter = new GcsMessageHandler(new GcsSessionFactory(gcs));
  outboundChannelAdapter.setRemoteDirectoryExpression(new ValueExpression<>("your-gcs-bucket"));

  return outboundChannelAdapter;
}

Sample

A sample application is available.