GH-9215: Honor back-pressure in FluxMessageChannel

Fixes: #9215

* Instead of `share()` use `publish(1).refCount()` to prefetch only item from upstream.
* Also remove `publishOn(this.scheduler)` for upstream publishers in favor of opt-in on the consumer side.

**Auto-cherry-pick to `6.3.x` & `6.2.x`**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -28,8 +28,6 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
-import reactor.core.scheduler.Scheduler;
-import reactor.core.scheduler.Schedulers;
 import reactor.util.context.ContextView;
 
 import org.springframework.core.log.LogMessage;
@@ -54,8 +52,6 @@
 public class FluxMessageChannel extends AbstractMessageChannel
 		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
 
-	private final Scheduler scheduler = Schedulers.boundedElastic();
-
 	private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
 
 	private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
@@ -107,7 +103,8 @@ private boolean tryEmitMessage(Message<?> message) {
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
 		this.sink.asFlux()
 				.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0))
-				.share()
+				.publish(1)
+				.refCount()
 				.subscribe(subscriber);
 
 		Mono<Boolean> subscribersBarrier =
@@ -148,7 +145,7 @@ public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 		Flux<Object> upstreamPublisher =
 				Flux.from(publisher)
 						.delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next())
-						.publishOn(this.scheduler)
+//						.publishOn(this.scheduler)
 						.flatMap((message) ->
 								Mono.just(message)
 										.handle((messageToHandle, syncSink) -> sendReactiveMessage(messageToHandle))
@@ -185,7 +182,6 @@ public void destroy() {
 		this.upstreamSubscriptions.dispose();
 		this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
 		this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
-		this.scheduler.dispose();
 		super.destroy();
 	}