Solace Spring Cloud Stream Function (Processor pattern) with StreamBridge

seph

Hi, I currently have a usecase wherein I have implemented a Processor pattern using the typical Bean Function<?,?> for my SCS application. I understand that a Processor pattern is a straightforward Input-Output Solace stream. But now I have an additional requirement that, within the processor, an arbitrary event message needs to emit another Solace Topic. An example of this arbitrary event is an application exception that must be emitted and sent to another service via Solace Topic.

I researched and found StreamBridge to be the current way to implement such arbitrary message sending. But when I tried doing so (add new producer config i.e. "solace.bindings.<binding-name>.producer.<queueNameExpression, etc.>", when streamBridge.send() is hit in SCS Processor flow, the message still gets sent to the consumer queue.

Previous style of implementations I saw for such usecase involve @StreamListener which as I've read is now a deprecated SCS feature so I wouldn't try that out anymore.

With this, I'm not sure whether it is really possible to combine both SCS Processor and Streambridge together. If anyone has ideas, I'll appreciate any thoughts towards my above usecase. Thanks!

marc

Hi @seph,

With StreamBridge you don't actually have to pre-configure the binding ahead of time. So within your processor which has it's regular input/output bindings you can just say streamBridge.send(topic, payload);

See if that gives you what you need :)

Here is an example using StreamBridge with a Consumer function, but it should work just the same inside of a Function one: https://github.com/SolaceSamples/solace-samples-spring/blob/master/cloud-stream-dynamic-destination-processor/src/main/java/com/solace/samples/spring/scs/DynamicDestinationProcessor.java#L73:L81

seph

Hi @marc,

Thanks a lot for the insights. Really helped establishing a baseline implementation. I was able to determine the actual behavior that was happening in my application through thorough debugs.

First, it is indeed possible to use Function<?,?> and StreamBridge together.

Next I've noticed is that only when the application throws an Exception within the Stream, is when the message is sent back to consumer queue. I'll have a look into the Error Handling for SCS and Solace specific implementation for this.

Lastly, the Exception encountered is happening within the implementation for StreamBridge. See https://github.com/spring-cloud/spring-cloud-stream/blob/main/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java#L201-L203

Wherein I was using streamBridge.send("bindingName-out-0", java-object). To resolve, I resorted to convert java object to byte array before sending.