An implementation of Spring's Cloud Stream Binder for integrating with Solace PubSub+ message brokers. The Spring Cloud Stream Binder project provides a higher-level abstraction towards messaging that standardizes the development of distributed message-based systems.

• Overview
• Spring Cloud Stream Binder
• Using it in your Application
• Configuration Options
• Failed Message Error Handling
• Building the Project Yourself
• Contributing
• Authors
• License
• Resources

The Solace implementation of the Spring Cloud Stream Binder maps the following concepts from Spring to Solace:

• Destinations to topic subscriptions
• Consumer groups to durable queues
• Anonymous consumer groups to temporary queues

And internally, each consumer group queue is subscribed to at least their destination topic. So a typical message flow would then appear as follows:

1. Producer bindings publish messages to their destination topics
2. Consumer group queues receive the messages published to their destination topic
3. Consumers of a particular consumer group consume messages from their group in a round-robin fashion (by default)

Note that partitioning is not yet supported by this version of the binder.

Also, it will be assumed that you have a basic understanding of the Spring Cloud Stream project. If not, then please refer to Spring's documentation. For the sake of brevity, this document will solely focus on discussing components unique to Solace.

This project extends the Spring Cloud Stream Binder project. If you are new to Spring Cloud Stream, check out their documentation.

The following is a brief excerpt from that document:

Spring Cloud Stream is a framework for building message-driven microservice applications. Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers. It provides opinionated configuration of middleware from several vendors, introducing the concepts of persistent publish-subscribe semantics, consumer groups, and partitions.

The releases from this project are hosted in Maven Central.

The easiest way to get started is to include the spring-cloud-starter-stream-solace in your application.

Here is how to include the spring cloud stream starter in your project using Gradle and Maven.

// Solace Spring Cloud Stream Binder
compile("com.solace.spring.cloud:spring-cloud-starter-stream-solace:1.0.+")

<!-- Solace Spring Cloud Stream Binder -->
<dependency>
  <groupId>com.solace.spring.cloud</groupId>
  <artifactId>spring-cloud-starter-stream-solace</artifactId>
  <version>1.0.+</version>
</dependency>

For a quick example of declaring a consumer binding in your application, take a look at Spring's introductory example.

Then for this example, an applicable Solace configuration file may look like:

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: testtock
          group: myconsumergroup

solace:
  java:
    host: tcp://192.168.133.64
    msgVpn: default
    clientUsername: default
    clientPassword: default
    connectRetries: -1
    reconnectRetries: -1

Notice that the latter half of this configuration actually originates from the JCSMP Spring Boot Auto-Configuration project.

Configuration of the Solace Spring Cloud Stream Binder is done through Spring Boot's externalized configuration. This is where users can control the binder's configuration options as well as the Solace Java API properties.

As for auto-configuration-related options required for auto-connecting to Solace message brokers, refer to the JCSMP Spring Boot Auto-Configuration documentation.

For general binder configuration options and properties, refer to the Spring Cloud Stream Reference Guide.

The following properties are available for Solace consumers only and must be prefixed with spring.cloud.stream.solace.bindings.<channelName>.consumer..

See SolaceConsumerProperties for the most updated list.

prefix

Naming prefix for all topics and queues.

Default: ""

queueAccessType

Access type for the consumer group queue.

Default: EndpointProperties.ACCESSTYPE_NONEXCLUSIVE

queuePermission

Permissions for the consumer group queue.

Default: EndpointProperties.PERMISSION_CONSUME

queueDiscardBehaviour

If specified, whether to notify sender if a message fails to be enqueued to the consumer group queue.

Default: null

queueMaxMsgRedelivery

Sets the maximum message redelivery count on consumer group queue. (Zero means retry forever).

Default: null

queueMaxMsgSize

Maximum message size for the consumer group queue.

Default: null

queueQuota

Message spool quota for the consumer group queue.

Default: null

queueRespectsMsgTtl

Whether the consumer group queue respects Message TTL.

Default: null

queueAdditionalSubscriptions

An array of additional topic subscriptions to be applied on the consumer group queue.

These subscriptions may also contain wildcards.

The prefix property is not applied on these subscriptions.

Default: String[0]

anonymousGroupPostfix

Naming postfix for the anonymous consumer group queue.

Default: "anon"

polledConsumerWaitTimeInMillis

Rate at which polled consumers will receive messages from their consumer group queue.

Default: 100

requeueRejected

Whether message processing failures should be re-queued when autoBindDmq is false and after all binder-internal retries have been exhausted.

Default: false

autoBindDmq

Whether to automatically create a durable dead message queue to which messages will be republished when message processing failures are encountered. Only applies once all internal retries have been exhausted.

Default: false

dmqAccessType

Access type for the DMQ.

Default: EndpointProperties.ACCESSTYPE_NONEXCLUSIVE

dmqPermission

Permissions for the DMQ.

Default: EndpointProperties.PERMISSION_CONSUME

dmqDiscardBehaviour

If specified, whether to notify sender if a message fails to be enqueued to the DMQ.

Default: null

dmqMaxMsgRedelivery

Sets the maximum message redelivery count on the DMQ. (Zero means retry forever).

Default: null

dmqMaxMsgSize

Maximum message size for the DMQ.

Default: null

dmqQuota

Message spool quota for the DMQ.

Default: null

dmqRespectsMsgTtl

Whether the DMQ respects Message TTL.

Default: null

republishedMsgTtl

The number of milliseconds before republished messages are discarded or moved to a Solace-internal Dead Message Queue.

Default: null

The following properties are available for Solace producers only and must be prefixed with spring.cloud.stream.solace.bindings.<channelName>.producer..

See SolaceProducerProperties for the most updated list.

prefix

Naming prefix for all topics and queues.

Default: ""

queueAccessType

Access type for the required consumer group queue.

Default: EndpointProperties.ACCESSTYPE_NONEXCLUSIVE

queuePermission

Permissions for the required consumer group queue.

Default: EndpointProperties.PERMISSION_CONSUME

queueDiscardBehaviour

If specified, whether to notify sender if a message fails to be enqueued to the required consumer group queue.

Default: null

queueMaxMsgRedelivery

Sets the maximum message redelivery count on the required consumer group queue. (Zero means retry forever).

Default: null

queueMaxMsgSize

Maximum message size for the required consumer group queue.

Default: null

queueQuota

Message spool quota for the required consumer group queue.

Default: null

queueRespectsMsgTtl

Whether the required consumer group queue respects Message TTL.

Default: null

queueAdditionalSubscriptions

A mapping of required consumer groups to arrays of additional topic subscriptions to be applied on each consumer group's queue.

These subscriptions may also contain wildcards.

The prefix property is not applied on these subscriptions.

Default: Empty Map<String,String[]>

msgTtl

The number of milliseconds before messages are discarded or moved to a Solace-internal Dead Message Queue.

Default: null

msgInternalDmqEligible

The DMQ here is not those which the binder creates when autoBindDmq is enabled, but instead, refers to the DMQ defined by the Solace message broker itself.

Default: false

Spring cloud stream binders already provides a number of application-internal reprocessing strategies for failed messages during message consumption such as:

• Forwarding errors to various Spring error message channels
• Internally re-processing the failed messages through the usage of a retry template

However, after all internal error handling strategies have been exhausted, the Solace implementation of the binder would by default reject messages that consumer bindings fail to process. Though it may be desirable for these failed messages be preserved and externally re-processed, in which case this binder also provides 2 error handling strategies that consumer bindings can be configured to use:

• Re-queuing the message onto that consumer group's queue
• Re-publishing the message to another queue (a dead-message queue) for some external application/binding to process

Note that both strategies cannot be used at the same time, and that enabling both of them would result in the binder treating it as if message re-queuing was disabled. That is to say, re-publishing failed messages to a dead-message queues supersedes message re-queuing.

A simple error handling strategy in which failed messages are simply re-queued onto the consumer group's queue. This is very similar to simply enabling the retry template (setting maxAttempts to a value greater than 1), but allows for the failed messages to be re-processed by the message broker.

First, it must be noted that the dead message queue (DMQ) that will be discussed in this section is different from the regular Solace DMQ. In particular, the standard Solace DMQ is used for re-routing failed messages as a consequence of Solace PubSub+ messaging features such as TTL expiration or exceeding a message's max redelivery count. Whereas the purpose of a Solace binder DMQ is for re-routing messages which had been successfully consumed from the message broker, yet cannot be processed by the binder. For simplicity, in this document all mentions of the "DMQ" refers to the Solace binder DMQ.

A DMQ can be provisioned for a particular consumer group by setting the autoBindDmq consumer property to true. This DMQ is simply another durable queue which, aside from its purpose, is not much from the queue provisioned for consumer groups. These DMQs are named using a period-delimited concatenation of their consumer group name and "dmq". And like the queue used for consumer groups, their endpoint properties can be configured by means of any consumer properties whose names begin with "dmq".

Note that DMQs are not intended to be used with anonymous consumer groups. Since the names of these consumer groups, and in turn the name of their would-be DMQs, are randomly generated at runtime, it would provide little value to create bindings to these DMQs because of their unpredictable naming and temporary existence.

This project depends on maven for building. To build the jar locally, check out the project and build from source by doing the following:

git clone https://github.com/SolaceProducts/spring-cloud-stream-binder-solace.git
cd spring-cloud-stream-binder-solace
mvn package

This will build the Solace binder core, the Solace binder itself, and the Solace spring cloud stream starter jars.

Note: As currently setup, the build requires Java 1.8. If you want to use another older version of Java adjust the build accordingly.

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.

See the list of contributors who participated in this project.

This project is licensed under the Apache License, Version 2.0. - See the LICENSE file for details.

For more information about Spring Cloud Streams try these resources:

• Spring Docs - Spring Cloud Stream Reference Guide
• GitHub Samples - Spring Cloud Stream Sample Applications
• Github Source - Spring Cloud Stream Source Code

For more information about Solace technology in general please visit these resources:

• The Solace Developer Portal website at: http://dev.solace.com
• Understanding Solace technology
• Ask the Solace community