Spring Integration 5.0 Reactor types support

Question

From release notes (https://spring.io/blog/2017/11/29/spring-integration-5-0-ga-available):

Reactive Streams support via FluxMessageChannel, ReactiveStreamsConsumer and direct org.reactivestreams.Subscriber implementation in the AbstractMessageHandler;

My understanding for Reactor support was e.g. you can return Mono/Flux from a transformer/handler, and Spring Integration will automatically transform it to Messages while respecting back pressure. Unfortunately, I cannot make it work like that, e.g.:

IntegrationFlows.from("input")
                .handle((p, h) -> Flux.just(1, 2, 3))
                .log("l1")
                .channel("output")
                .get();

still logs one Message with FluxArray typed payload instead of three Messages with Integer payloads.

2017-12-18 17:12:33.262  INFO 97471 --- [nio-8080-exec-1] l1 : GenericMessage [payload=FluxArray, headers={id=a9701681-9945-f953-8b72-df369c2982a3, timestamp=1513613553262}]

Also, there is nothing in docs according this behaviour and new

FluxMessageChannel, ReactiveStreamsConsumer and direct org.reactivestreams.Subscriber implementation in the AbstractMessageHandler

So my question is, do I understand implemented Reactor support correctly, and where can I find any info on that topic?

Artem Bilan Artem Bilan · Accepted Answer · 2017-12-18T16:32:31

Since we here in messaging and that really doesn't matter for the message what kind of payload you return from your service, everything is just wrapped to the Message as is. You need a special component to understand this payload. One of them is Splitter. This one determines that your payload is a Reactive Streams Publisher and iterated over that as a Flux.

Another component is WebFluxInboundEndpoint which supports this kind of payloads natively.

Your custom Service Activator might expect Flux as an argument to deal with.

But nothing happens automatically. Spring Integration supports Reactive types, but doesn't do they processing without end-user preferences.

BTW, the splitter should be supplied with the FluxMessageChannel as an output to process the splitted Flux via back--pressure manner.

Feel free to raise A JIRA about documenting FluxMessageChannel. Indeed we have missed that. The ReactiveStreamsConsumer needs more love as well and we have some plans for 5.1 to improve Reactive Streams model and we'll try to make it more flexible or even like an option to turn on it by default. Nothing can promise from today though.

Spring Integration 5.0 Reactor types support

1 Answers