Add toReactivePublisher(autoStartOnSubscribe) (#3657)

* Add `toReactivePublisher(autoStartOnSubscribe)`

The `IntegrationFlowBuilder.toReactivePublisher()` returns a `Publisher<Message<?>>`
which may be subscribed somewhere late in the application logic, e.g.
when WebSocket (or RSocket) subscription is initiated by the external client.
In between application context startup and that subscription moment, the `IntegrationFlow`
must not try to produce messages since there is nothing to consumer them from the
`Publisher<Message<?>>` side.
One of the way is to have a source endpoint not started automatically and control its
lifecycle from the point fo reactive subscription

* Introduce an `IntegrationFlowBuilder.toReactivePublisher(boolean autoStartOnSubscribe)`
to let the framework do a job for an `IntegrationFlow` lifecycle control.
This way end-user doesn't need to know autowire a starting endpoint and use `doOnSubscribe()`
and similar callbacks
* Change `ConsumerEndpointFactoryBean` log message about a `FixedSubscriberChannel`
to `INFO` since an `autoStartup = false` really does not have any effect and there is nothing
for end-user to worry about.
The `IntegrationFlow` knows nothing about each endpoint internals and cannot control which
mark as `autoStartup = false` and which not

* Fix languge in JavaDocs

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

@@ -374,7 +374,7 @@ public class ConsumerEndpointFactoryBean
 						+ "', since '" + channel + "' is a SubscribableChannel (not pollable).");
 		this.endpoint = new EventDrivenConsumer((SubscribableChannel) channel, this.handler);
 		if (Boolean.FALSE.equals(this.autoStartup) && channel instanceof FixedSubscriberChannel) {
-			LOGGER.warn("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel");
+			LOGGER.info("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel");
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java

@@ -2934,8 +2934,21 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 	 * @param <T> the expected {@code payload} type
 	 * @return the Reactive Streams {@link Publisher}
 	 */
-	@SuppressWarnings(UNCHECKED)
 	protected <T> Publisher<Message<T>> toReactivePublisher() {
+		return toReactivePublisher(false);
+	}
+
+	/**
+	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
+	 * @param autoStartOnSubscribe start message production and consumption in the flow,
+	 * when a subscription to the publisher is initiated.
+	 * If this set to true, the flow is marked to not start automatically by the application context.
+	 * @param <T> the expected {@code payload} type
+	 * @return the Reactive Streams {@link Publisher}
+	 * @since 5.5.6
+	 */
+	@SuppressWarnings(UNCHECKED)
+	protected <T> Publisher<Message<T>> toReactivePublisher(boolean autoStartOnSubscribe) {
 		MessageChannel channelForPublisher = getCurrentMessageChannel();
 		Publisher<Message<T>> publisher;
 		Map<Object, String> components = getIntegrationComponents();
@@ -2959,7 +2972,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 
 		get();
 
-		return new PublisherIntegrationFlow<>(components, publisher);
+		return new PublisherIntegrationFlow<>(components, publisher, autoStartOnSubscribe);
 	}
 
 	protected <S extends ConsumerEndpointSpec<? super S, ? extends MessageHandler>> B register(S endpointSpec,

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -40,4 +40,9 @@ public final class IntegrationFlowBuilder extends IntegrationFlowDefinition<Inte
 		return super.toReactivePublisher();
 	}
 
+	@Override
+	public  <T> Publisher<Message<T>> toReactivePublisher(boolean autoStartOnSubscribe) { // NOSONAR
+		return super.toReactivePublisher(autoStartOnSubscribe);
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/dsl/PublisherIntegrationFlow.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,8 @@ import org.reactivestreams.Subscriber;
 
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.Flux;
+
 /**
  *
  * @param <T> the message payload type.
@@ -35,9 +37,26 @@ class PublisherIntegrationFlow<T> extends StandardIntegrationFlow implements Pub
 
 	private final Publisher<Message<T>> delegate;
 
-	PublisherIntegrationFlow(Map<Object, String> integrationComponents, Publisher<Message<T>> publisher) {
+	PublisherIntegrationFlow(Map<Object, String> integrationComponents, Publisher<Message<T>> publisher,
+			boolean autoStartOnSubscribe) {
+
 		super(integrationComponents);
-		this.delegate = publisher;
+		Flux<Message<T>> flux =
+				Flux.from(publisher)
+						.doOnCancel(this::stop)
+						.doOnTerminate(this::stop);
+
+		if (autoStartOnSubscribe) {
+			flux = flux.doOnSubscribe((sub) -> start());
+			for (Object component : integrationComponents.keySet()) {
+				if (component instanceof EndpointSpec) {
+					((EndpointSpec<?, ?, ?>) component).autoStartup(false);
+				}
+			}
+		}
+
+		this.delegate = flux;
+
 	}
 
 	@Override

spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java

@@ -54,6 +54,7 @@ import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.scheduler.Schedulers;
 
@@ -104,19 +105,22 @@ public class ReactiveStreamsTests {
 
 	@Test
 	void testReactiveFlow() throws Exception {
+		assertThat(this.messageSource.isRunning()).isFalse();
 		List<String> results = new ArrayList<>();
 		CountDownLatch latch = new CountDownLatch(6);
-		Flux.from(this.publisher)
+		Disposable disposable =
+				Flux.from(this.publisher)
 				.map(m -> m.getPayload().toUpperCase())
 				.subscribe(p -> {
 					results.add(p);
 					latch.countDown();
 				});
-		this.messageSource.start();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		String[] strings = results.toArray(new String[0]);
 		assertThat(strings).isEqualTo(new String[]{ "A", "B", "C", "D", "E", "F" });
-		this.messageSource.stop();
+
+		disposable.dispose();
+		assertThat(this.messageSource.isRunning()).isFalse();
 	}
 
 	@Test
@@ -249,11 +253,10 @@ public class ReactiveStreamsTests {
 			return IntegrationFlows
 					.from(() -> new GenericMessage<>("a,b,c,d,e,f"),
 							e -> e.poller(p -> p.trigger(ctx -> this.invoked.getAndSet(true) ? null : new Date()))
-									.autoStartup(false)
 									.id("reactiveStreamsMessageSource"))
 					.split(String.class, p -> p.split(","))
 					.log()
-					.toReactivePublisher();
+					.toReactivePublisher(true);
 		}
 
 		@Bean