Reactive adapter adapter [PoC Only Do not merge]

[garyrussell]

No description provided.

[bsideup]

it is not recommended to leak emitter. Use one of the Processors:
https://projectreactor.io/docs/core/release/reference/#processors

[artembilan]

Actually we've been advised exactly opposite: the processors are going to go away from Reactor Core, so would be better to rely on the similar structure provided by the Flux.create().

Anyway this PR is too old and really deserves revisiting one day.

I hope you'll come up soon with the Flux.fromQueue() as we discussed recently in call, so we would have a natural back-pressure for the constantly emitting source.

Thank you for feedback, @bsideup , anyway!

[bsideup]

They are not going away, this is a misunderstanding. Flux.fromQueue won't give you backpressure I think, the processes is the way to go :)

[artembilan]

Yeah... Let's reconsider that in the next milestone!
We have a lot of similar Flux.create() here in Spring Integration.

I'll ping you later if needed.

Thank you, Sergey, again!

[bsideup]

np! Feel free to assign me as a reviewer, or ask for more info directly 👍

[artembilan]

With the MessageProducerSupport.subscribeToPublisher(Publisher<? extends Message<?>>) and FluxMessageChannel combination introduced in the #3240 we should think about closing this PR.
The more I look into this the more I'm sure such a channel bypassing approach fully breaks the Spring Integration purpose.
If you definitely would like to handle as a Publisher, it would be better to inject (currently) a FluxMessageChannel into your service and process it respectively using Reactor or RxJava operators.
This way we still keep a loosely-coupled principle between endpoint and would have all the tracking and metrics information on Spring Integration graph.

We definitely may think about a @ServiceActivator + Publsiher method parameter approach against an input channel (independently Flux it or not), but seems for me that may be possible already after introducing a @Reactive marker on that annotation (or method level like an @EndpointId). See #2031 .

Thoughts?

Thanks

[artembilan]

Another argument against this PR is with Java DSL's:

	/**
	 * 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) {

[artembilan]

Superseded with #3503