Rely on MProducerSupport.active for Flux (#3423)

* Rely on `MProducerSupport.active` for `Flux`

* Fix `MessageProducerSupport` to extract an `active` flag and set it before
`isRunning` - the `Flux` subscription relies on the `takeWhile()`
where in case of `autoStartup = false` we will never start consume because
it is set to `true` already after `doStart()`
* Refactor all the `MessageProducerSupport` implementation with similar
`active` state to use already one from the super class

**Cherry-pick to 5.3.x**

* * Remove `MessageProducerSupport.setActive()`
to not let to mutate it from the implementations
* Set `active` to `false` in the `destroy()`
* Clean up and fix typos in the affected `JmsMessageDrivenEndpoint`

* * Pull `active` flag down to the `AbstractEndpoint`
* Set `active = true` in the `start()` before calling `doStart()`
* Do same for `active = false` in the `stop()`
* Clean up `AbstractEndpoint` impls to not call `doStart/doStop` for nothing
* Refactor endpoints to rely on the `active` state from the `AbstractEndpoint`
not their own

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java

@@ -47,14 +47,6 @@
 public abstract class AbstractEndpoint extends IntegrationObjectSupport
 		implements SmartLifecycle, DisposableBean {
 
-	private boolean autoStartupSetExplicitly;
-
-	private volatile boolean autoStartup = true;
-
-	private volatile int phase = 0;
-
-	private volatile boolean running;
-
 	protected final ReentrantLock lifecycleLock = new ReentrantLock(); // NOSONAR
 
 	protected final Condition lifecycleCondition = this.lifecycleLock.newCondition(); // NOSONAR
@@ -63,6 +55,16 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport
 
 	private SmartLifecycleRoleController roleController;
 
+	private boolean autoStartup = true;
+
+	private boolean autoStartupSetExplicitly;
+
+	private int phase = 0;
+
+	private volatile boolean running;
+
+	private volatile boolean active;
+
 	public void setAutoStartup(boolean autoStartup) {
 		this.autoStartup = autoStartup;
 		this.autoStartupSetExplicitly = true;
@@ -120,6 +122,7 @@ protected void onInit() {
 
 	@Override
 	public void destroy() {
+		stop();
 		if (this.roleController != null) {
 			this.roleController.removeLifecycle(this);
 		}
@@ -153,6 +156,7 @@ public final void start() {
 		this.lifecycleLock.lock();
 		try {
 			if (!this.running) {
+				this.active = true;
 				doStart();
 				this.running = true;
 				if (logger.isInfoEnabled()) {
@@ -170,6 +174,7 @@ public final void stop() {
 		this.lifecycleLock.lock();
 		try {
 			if (this.running) {
+				this.active = false;
 				doStop();
 				this.running = false;
 				if (logger.isInfoEnabled()) {
@@ -187,6 +192,7 @@ public final void stop(Runnable callback) {
 		this.lifecycleLock.lock();
 		try {
 			if (this.running) {
+				this.active = false;
 				doStop(callback);
 				this.running = false;
 				if (logger.isInfoEnabled()) {
@@ -211,6 +217,10 @@ protected void doStop(Runnable callback) {
 		callback.run();
 	}
 
+	public boolean isActive() {
+		return this.active;
+	}
+
 	/**
 	 * Subclasses must implement this method with the start behavior.
 	 * This method will be invoked while holding the {@link #lifecycleLock}.

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java

@@ -368,7 +368,7 @@ private Flux<Message<?>> createFluxGenerator() {
 																		triggerContext.lastActualExecutionTime(),
 																		new Date())
 												)), 1)
-				.repeat(this::isRunning)
+				.repeat(this::isActive)
 				.doOnSubscribe(subs -> this.subscription = subs);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -222,7 +222,7 @@ protected void subscribeToPublisher(Publisher<? extends Message<?>> publisher) {
 						.map(this::trackMessageIfAny)
 						.doOnComplete(this::stop)
 						.doOnCancel(this::stop)
-						.takeWhile((message) -> isRunning());
+						.takeWhile((message) -> isActive());
 
 		if (channelForSubscription instanceof ReactiveStreamsSubscribableChannel) {
 			((ReactiveStreamsSubscribableChannel) channelForSubscription).subscribeTo(messageFlux);

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java

@@ -1016,6 +1016,12 @@ private <T> T convert(Object source, Class<T> expectedReturnType) {
 		}
 	}
 
+	@Override
+	public void destroy() {
+		super.destroy();
+		this.gatewayMap.values().forEach(MethodInvocationGateway::destroy);
+	}
+
 	private static final class MethodInvocationGateway extends MessagingGatewaySupport {
 
 		private Expression receiveTimeoutExpression;

spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.endpoint;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import java.time.Duration;
 
 import org.junit.jupiter.api.Test;
 
@@ -50,17 +50,18 @@ public class ReactiveMessageProducerTests {
 
 	@Test
 	public void test() {
-		assertThat(this.producer.isRunning()).isTrue();
-
-		StepVerifier.create(
-				Flux.from(this.fluxMessageChannel)
-						.map(Message::getPayload)
-						.cast(String.class))
-				.expectNext("test1", "test2")
-				.thenCancel()
-				.verify();
-
-		assertThat(this.producer.isRunning()).isFalse();
+		StepVerifier stepVerifier =
+				StepVerifier.create(
+						Flux.from(this.fluxMessageChannel)
+								.map(Message::getPayload)
+								.cast(String.class))
+						.expectNext("test1", "test2")
+						.thenCancel()
+						.verifyLater();
+
+		this.producer.start();
+
+		stepVerifier.verify(Duration.ofSeconds(10));
 	}
 
 	@Configuration
@@ -79,10 +80,12 @@ public MessageProducerSupport producer() {
 
 						@Override
 						protected void doStart() {
+							super.doStart();
 							subscribeToPublisher(Flux.just("test1", "test2").map(GenericMessage::new));
 						}
 
 					};
+			producer.setAutoStartup(false);
 			producer.setOutputChannel(fluxMessageChannel());
 			return producer;
 		}

spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -41,6 +41,7 @@
  * @author Mark Fisher
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @see ApplicationEventMulticaster
  * @see ExpressionMessageProducerSupport
  */
@@ -51,8 +52,6 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
 
 	private ApplicationEventMulticaster applicationEventMulticaster;
 
-	private volatile boolean active;
-
 	private volatile long stoppedAt;
 
 	public ApplicationEventListeningMessageProducer() {
@@ -66,14 +65,13 @@ public ApplicationEventListeningMessageProducer() {
 	 * In addition, this method re-registers the current instance as a {@link GenericApplicationListener}
 	 * with the {@link ApplicationEventMulticaster} which clears the listener cache. The cache will be
 	 * refreshed on the next appropriate {@link ApplicationEvent}.
-	 *
 	 * @param eventTypes The event types.
 	 * @see ApplicationEventMulticaster#addApplicationListener
 	 * @see #supportsEventType
 	 */
 	public final void setEventTypes(Class<?>... eventTypes) {
 		Assert.notNull(eventTypes, "'eventTypes' must not be null");
-		Set<ResolvableType> eventSet = new HashSet<ResolvableType>(eventTypes.length);
+		Set<ResolvableType> eventSet = new HashSet<>(eventTypes.length);
 		for (Class<?> eventType : eventTypes) {
 			if (eventType != null) {
 				eventSet.add(ResolvableType.forClass(eventType));
@@ -94,7 +92,7 @@ public String getComponentType() {
 	@Override
 	protected void onInit() {
 		super.onInit();
-		this.applicationEventMulticaster = this.getBeanFactory()
+		this.applicationEventMulticaster = getBeanFactory()
 				.getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME,
 						ApplicationEventMulticaster.class);
 		Assert.notNull(this.applicationEventMulticaster,
@@ -104,21 +102,23 @@ protected void onInit() {
 
 	@Override
 	public void onApplicationEvent(ApplicationEvent event) {
-		if (this.active || ((event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent)
-				&& this.stoppedRecently())) {
-			if (event.getSource() instanceof Message<?>) {
-				this.sendMessage((Message<?>) event.getSource());
+		if (isActive() || ((event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent)
+				&& stoppedRecently())) {
+
+			Object source = event.getSource();
+			if (source instanceof Message<?>) {
+				sendMessage((Message<?>) source);
 			}
 			else {
-				Message<?> message = null;
+				Message<?> message;
 				Object result = extractObjectToSend(event);
 				if (result instanceof Message) {
 					message = (Message<?>) result;
 				}
 				else {
-					message = this.getMessageBuilderFactory().withPayload(result).build();
+					message = getMessageBuilderFactory().withPayload(result).build();
 				}
-				this.sendMessage(message);
+				sendMessage(message);
 			}
 		}
 	}
@@ -175,15 +175,10 @@ public int getOrder() {
 		return Ordered.LOWEST_PRECEDENCE;
 	}
 
-	@Override
-	protected void doStart() {
-		this.active = true;
-	}
-
 	@Override
 	protected void doStop() {
 		this.stoppedAt = System.currentTimeMillis();
-		this.active = false;
+		super.doStop();
 	}
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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,19 +23,21 @@
  * File tailer that delegates to the Apache Commons Tailer.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 3.0
  *
  */
 public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageProducerSupport
 		implements TailerListener {
 
-	private volatile Tailer tailer;
+	private long pollingDelay = 1000;
 
-	private volatile long pollingDelay = 1000;
+	private boolean end = true;
 
-	private volatile boolean end = true;
+	private boolean reopen = false;
 
-	private volatile boolean reopen = false;
+	private volatile Tailer tailer;
 
 	/**
 	 * The delay between checks of the file for new content in milliseconds.
@@ -71,8 +73,8 @@ public String getComponentType() {
 	@Override
 	protected void doStart() {
 		super.doStart();
-		Tailer theTailer = new Tailer(this.getFile(), this, this.pollingDelay, this.end, this.reopen);
-		this.getTaskExecutor().execute(theTailer);
+		Tailer theTailer = new Tailer(getFile(), this, this.pollingDelay, this.end, this.reopen);
+		getTaskExecutor().execute(theTailer);
 		this.tailer = theTailer;
 	}
 
@@ -88,9 +90,9 @@ public void init(Tailer tailer) {
 
 	@Override
 	public void fileNotFound() {
-		this.publish("File not found: " + this.getFile().getAbsolutePath());
+		publish("File not found: " + getFile().getAbsolutePath());
 		try {
-			Thread.sleep(this.getMissingFileDelay());
+			Thread.sleep(getMissingFileDelay());
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
@@ -99,7 +101,7 @@ public void fileNotFound() {
 
 	@Override
 	public void fileRotated() {
-		this.publish("File rotated: " + this.getFile().getAbsolutePath());
+		publish("File rotated: " + getFile().getAbsolutePath());
 	}
 
 	@Override
@@ -109,7 +111,7 @@ public void handle(String line) {
 
 	@Override
 	public void handle(Exception ex) {
-		this.publish(ex.getMessage());
+		publish(ex.getMessage());
 	}
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java

@@ -145,7 +145,6 @@ protected void publish(String message) {
 
 	@Override
 	protected void doStart() {
-		super.doStart();
 		if (this.idleEventInterval > 0) {
 			this.idleEventScheduledFuture = getTaskScheduler().scheduleWithFixedDelay(() -> {
 				long now = System.currentTimeMillis();
@@ -162,7 +161,6 @@ protected void doStart() {
 
 	@Override
 	protected void doStop() {
-		super.doStop();
 		if (this.idleEventScheduledFuture != null) {
 			this.idleEventScheduledFuture.cancel(true);
 		}