Unit Testing a Spring Cloud Stream Producer / Publisher

Now-a-days technology changes at such a pace, that the moment you take your foot off the gas, you start the catch up game. In my case, I took it off far too long and when I tried to start where I left off, the entire terrain appeared to be new even though things still had the same stability under the hood. Cutting to the chase, am talking about the updated Spring Cloud Stream functionality.

When I last coded and publisher using Spring Cloud Stream, it was pre-covid times. We used annotations like @Input and @Output to define the channels to be used for communicating with the binding implementations. And these channels would then be used to send and receive messages. To publish a message we would get the output channel from the declared channels and just send the message through.

But now, things have changed. Spring Cloud Stream has moved towards functional interfaces, so input and output channels have now been replaced by their functional counterparts Consumer and Supplier. In this article we will look at ways to create a publisher and also test the same using the provided test binder implementation.

Implementing a Publisher

A message publisher can be implemented in two ways:

Supplier Function

A Supplier function can be used if we would like to send a message every fixed interval called a PollingInterval. The supplier function can be coded as being imperative (as shown below) or reactive.

Polling interval can be specified by the property spring.cloud.stream.poller.fixedDelay and defaults to 1 second (1000 millis). There are other properties under spring.cloud.stream.poller to configure the nature of polling and fine tune it.

Stream Bridge

Often we would like to integrate the message publishing with an external trigger such as Rest API or a change in database record. In such conditions we need to use a StreamBridge.

In this case the messages are not auto published but depend on the invocation of publishMessage method.

Binding Mapping

By default spring will generate a binding name basis the method name. This is required only in case of Supplier as for the StreamBridge the binding is passed in as a parameter. The default naming convention followed by spring is <method-name> + -out- + index. The index value is typically 0 (single output) and is more relevant for cases where multiple inputs and outputs are desired.

So for our supplier based publisher, the binding name (generate by default) becomes

stringSupplier-out-0

In most of the cases this would suffice. But if there are constraints on the binding name to be used, this default generated value for binding can be overriden using the spring property spring.cloud.stream.function.bindings.binding-name as follows:

From this point onwards, the binding name test-out can be used to define properties in the properties file instead of the default stringSupplier-out-0.

Destination Mapping

The binding name can be used to define several properties in the application.properties file. The most prominent one being destination which will ascertain the queue or exchange (in case of RabbitMQ) the messages will be delivered to.

In the above configuration the binding test-out has been bound to the destination (on the actual broker) with name test-detination.

Testing

For testing the publisher, or for that matter any feature of Spring Cloud Stream, we can use the test framework provided by the SCS team. The new test framework is capable of simulating real binder scenarios which were not available in the previous implementations.

Dependency Resolution

In order to include the test framework use the following maven/gradle dependency:

Testing Spring Bridge Based Publisher

Class-Level Configuration

Once included there are two ways to code the test. One way is to import the TestChannelBinderConfiguration:

When using the TestChannelBinderConfiguration spring automatically injects both an InputDestination and an OutputDestination. These can be used in conjunction with our publishers and subscribers.

In our case since we are testing our publisher, we will use the OutputDestination to subscribe to the channel on which we publish the messages so that we can use assertions on the messages received.

Message<byte[]> result = outputDestination.receive(100, "test-destination");        
assertThat(result).isNotNull();        
assertThat(new String(result.getPayload())).isEqualTo("Hello");

Note: We are listening on the destination test-destination which has been defined as the destination for the binding test-out where the messages are being sent.

Method-Level Configuration

Alternatively, if we do not want to create a class specifically for these tests, the entire setup can be done within a method by initializing the spring context with appropriate input params

The relevant properties defined in the application.properties file (under java source) need to be copied over to the application.properties file (under the test source) so that the tests can use them for setup.

Testing Supplier Function Based Publisher

In the case of Suplier function based publisher, we don’t require to publish a message as a part of the test case as it will be automatically published by the framework. All we need to do is to listen for the published messages at the right interval:

The output destination will receive messages at a fixed delay of 1 second (unless overriden in the properties as described above). Therefore testing for new messages with appropriate delay would result in a successful outcome.

Important: A very important thing while implementing Supplier function is to set the property spring.cloud.function.definition in the application.properties file. This property can be ignored if there is only one supplier or consumer in the entire project, but becomes absolutely mandatory in case there are more than one. In the absence of this property (for cases with multiple supplier, consumer or combinations of both), spring will not generate messages as expected.