Feature Request - handleReactive(ReactiveMessageHandler)

[jifeiiii]

Expected Behavior
I'd like there to be a reactive handleReactive(ReactiveMessageHandler) operator that will enable me to easily write reactive code as a handler.

Current Behavior
There is a need to use the ReactiveMessageHandlerAdapter class and write the code inside of it.

Context
The workaround is pretty simple, but an implementation of such a function seems to be pretty simple either. Here's a SO question to support the issue.

[artembilan]

Just to clarify: such an operator indeed going to be based on the ReactiveMessageHandler only, which could be represent as Lambda for simple use-cases.
But most important it is going to be terminal operator with an exact meaning that there is no way to continue a flow after it.
Something similar to the IntegrationFlow nullChannel() or IntegrationFlow to(IntegrationFlow other).

[jifeiiii]

I also think that it would be useful to let the user choose how exactly to subscribe, and to manage publishOn easily

[artembilan]

The publishOn() is a part of the reactive() endpoint configuration. It has nothing to do with particular handler.
I'm not sure what kind of choice you'd like to see for the subscription. This is going to be the end of the flow.
Therefore Mono.then() is the way to go in the particular ReactiveMessageHandler impl: the framework will do the rest for you.

[markusherbert]

@artembilan I think this issue talks about a correct point.. You currently can not access a flux and end the flow through it by yourself. This is really a blocker when it comes to developing custom outbound channel adapters or just custom reactive code. In this manner, I think Reactive Spring Integration is very problematic. The solution may be to create a handleFlux(Flux<Message<?>>). I think this is what lots of people search for (or me at least).

In my point of view, if Spring Integration enables this, it can easily become the 'project reactor framework' - a thing that lots of people would use happily.
Such handleFlux() is pretty simple to add, and can add lots of value for Reactive Spring Integration users (that currently need to do instanceof Flux inside the message handler class to perform flux operations.

[markusherbert]

In general, @artembilan, is it currently possible to subscribe manually? I think a built-in feature for this can provide a lot of flexibility to the reactive spring integration users

[artembilan]

There is already something like handleFlux() in Spring Integration Java DSL. It is called fluxTransform():

	/**
	 * Populate a {@link FluxMessageChannel} to start a reactive processing for upstream data,
	 * wrap it to a {@link Flux}, apply provided {@link Function} via {@link Flux#transform(Function)}
	 * and emit the result to one more {@link FluxMessageChannel}, subscribed in the downstream flow.
	 * @param fluxFunction the {@link Function} to process data reactive manner.
	 * @param <I> the input payload type.
	 * @param <O> the output type.
	 * @return the current {@link BaseIntegrationFlowDefinition}.
	 */
	@SuppressWarnings(UNCHECKED)
	public <I, O> B fluxTransform(Function<? super Flux<Message<I>>, ? extends Publisher<O>> fluxFunction) {

I even mentioned this in the SO you are pointing to.
I'm not sure what else we need...

This particular issue is exactly for the ReactiveMessageHandler convenience which can be used only in the end of flow.
So, let's don't hijack it and talk only about specific task in a single GH issue!