Improve ImapIdleChannelAdapter (#3045)

* Improve ImapIdleChannelAdapter

* We should not destroy a `TaskExecutor` in the `stop()`, especially
when we are going to restart eventually.
Move that logic into `destroy()`
* we should not destroy `MailReceiver` in the `stop()`; we don't
reinstate it in the `start()`.
Move the logic into `destroy()`
* Wrap `ReceivingTask` and `IdleTask` into `isRunning()` condition to
avoid task executions when we are in stopped state
* Remove `ImapMailReceiverTests.testExecShutdown()` since it is not
relevant any more and doesn't reflect `mail` module requirements

* * Add `ImapMailReceiver.cancelPing()` hook
* Also close folder for each `stop()`, as well as in the `destroy()`

spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java

@@ -390,7 +390,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		}
 	}
 
-	private void closeFolder() {
+	protected void closeFolder() {
 		this.folderReadLock.lock();
 		try {
 			MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);

spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java

@@ -78,7 +78,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 
 	private boolean shouldReconnectAutomatically = true;
 
-	private Executor sendingTaskExecutor;
+	private Executor sendingTaskExecutor = Executors.newFixedThreadPool(1);
 
 	private boolean sendingTaskExecutorSet;
 
@@ -95,6 +95,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 
 	public void setTransactionSynchronizationFactory(
 			TransactionSynchronizationFactory transactionSynchronizationFactory) {
+
 		this.transactionSynchronizationFactory = transactionSynchronizationFactory;
 	}
 
@@ -102,7 +103,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 		this.adviceChain = adviceChain;
 	}
 
-
 	/**
 	 * Specify an {@link Executor} used to send messages received by the
 	 * adapter.
@@ -156,9 +156,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 	protected void doStart() {
 		TaskScheduler scheduler = getTaskScheduler();
 		Assert.notNull(scheduler, "'taskScheduler' must not be null");
-		if (this.sendingTaskExecutor == null) {
-			this.sendingTaskExecutor = Executors.newFixedThreadPool(1);
-		}
 		this.receivingTask = scheduler.schedule(new ReceivingTask(), this.receivingTaskTrigger);
 	}
 
@@ -166,19 +163,16 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 	// guarded by super#lifecycleLock
 	protected void doStop() {
 		this.receivingTask.cancel(true);
-		try {
-			this.mailReceiver.destroy();
-		}
-		catch (Exception e) {
-			throw new IllegalStateException(
-					"Failure during the destruction of Mail receiver: " + this.mailReceiver, e);
-		}
-		/*
-		 * If we're running with the default executor, shut it down.
-		 */
+		this.mailReceiver.cancelPing();
+	}
+
+	@Override
+	public void destroy() {
+		super.destroy();
+		this.mailReceiver.destroy();
+		// If we're running with the default executor, shut it down.
 		if (!this.sendingTaskExecutorSet && this.sendingTaskExecutor != null) {
 			((ExecutorService) this.sendingTaskExecutor).shutdown();
-			this.sendingTaskExecutor = null;
 		}
 	}
 
@@ -250,17 +244,19 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 
 		@Override
 		public void run() {
-			try {
-				ImapIdleChannelAdapter.this.idleTask.run();
-				logger.debug("Task completed successfully. Re-scheduling it again right away.");
-			}
-			catch (Exception e) { //run again after a delay
-				if (logger.isWarnEnabled()) {
-					logger.warn("Failed to execute IDLE task. Will attempt to resubmit in "
-							+ ImapIdleChannelAdapter.this.reconnectDelay + " milliseconds.", e);
+			if (isRunning()) {
+				try {
+					ImapIdleChannelAdapter.this.idleTask.run();
+					logger.debug("Task completed successfully. Re-scheduling it again right away.");
+				}
+				catch (Exception e) { //run again after a delay
+					if (logger.isWarnEnabled()) {
+						logger.warn("Failed to execute IDLE task. Will attempt to resubmit in "
+								+ ImapIdleChannelAdapter.this.reconnectDelay + " milliseconds.", e);
+					}
+					ImapIdleChannelAdapter.this.receivingTaskTrigger.delayNextExecution();
+					publishException(e);
 				}
-				ImapIdleChannelAdapter.this.receivingTaskTrigger.delayNextExecution();
-				publishException(e);
 			}
 		}
 
@@ -275,38 +271,33 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 
 		@Override
 		public void run() {
-			final TaskScheduler scheduler = getTaskScheduler();
-			Assert.notNull(scheduler, "'taskScheduler' must not be null");
-			/*
-			 * The following shouldn't be necessary because doStart() will have ensured we have
-			 * one. But, just in case...
-			 */
-			Assert.state(ImapIdleChannelAdapter.this.sendingTaskExecutor != null,
-					"'sendingTaskExecutor' must not be null");
-
-			try {
-				logger.debug("waiting for mail");
-				ImapIdleChannelAdapter.this.mailReceiver.waitForNewMessages();
-				Folder folder = ImapIdleChannelAdapter.this.mailReceiver.getFolder();
-				if (folder != null && folder.isOpen()) {
-					Object[] mailMessages = ImapIdleChannelAdapter.this.mailReceiver.receive();
-					if (logger.isDebugEnabled()) {
-						logger.debug("received " + mailMessages.length + " mail messages");
-					}
-					for (Object mailMessage : mailMessages) {
-						Runnable messageSendingTask = createMessageSendingTask(mailMessage);
-						ImapIdleChannelAdapter.this.sendingTaskExecutor.execute(messageSendingTask);
+			if (isRunning()) {
+				try {
+					logger.debug("waiting for mail");
+					ImapIdleChannelAdapter.this.mailReceiver.waitForNewMessages();
+					Folder folder = ImapIdleChannelAdapter.this.mailReceiver.getFolder();
+					if (folder != null && folder.isOpen() && isRunning()) {
+						Object[] mailMessages = ImapIdleChannelAdapter.this.mailReceiver.receive();
+						if (logger.isDebugEnabled()) {
+							logger.debug("received " + mailMessages.length + " mail messages");
+						}
+						for (Object mailMessage : mailMessages) {
+							Runnable messageSendingTask = createMessageSendingTask(mailMessage);
+							if (isRunning()) {
+								ImapIdleChannelAdapter.this.sendingTaskExecutor.execute(messageSendingTask);
+							}
+						}
 					}
 				}
-			}
-			catch (MessagingException e) {
-				logger.warn("error occurred in idle task", e);
-				if (ImapIdleChannelAdapter.this.shouldReconnectAutomatically) {
-					throw new IllegalStateException("Failure in 'idle' task. Will resubmit.", e);
-				}
-				else {
-					throw new org.springframework.messaging.MessagingException(
-							"Failure in 'idle' task. Will NOT resubmit.", e);
+				catch (MessagingException e) {
+					logger.warn("error occurred in idle task", e);
+					if (ImapIdleChannelAdapter.this.shouldReconnectAutomatically) {
+						throw new IllegalStateException("Failure in 'idle' task. Will resubmit.", e);
+					}
+					else {
+						throw new org.springframework.messaging.MessagingException(
+								"Failure in 'idle' task. Will NOT resubmit.", e);
+					}
 				}
 			}
 		}

spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java

@@ -156,9 +156,20 @@ public class ImapMailReceiver extends AbstractMailReceiver {
 		if (this.isInternalScheduler) {
 			((ThreadPoolTaskScheduler) this.scheduler).shutdown();
 		}
+		cancelPing();
+	}
+
+	/**
+	 * The hook to be called when we need to cancel the current ping task and close the mail folder.
+	 * In other words: when IMAP idle should be stopped for some reason.
+	 * The next {@link #waitForNewMessages()} call will re-open the folder and start a new ping task.
+	 * @since 5.2
+	 */
+	public void cancelPing() {
 		if (this.pingTask != null) {
 			this.pingTask.cancel(true);
 		}
+		closeFolder();
 	}
 
 	/**

spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java

@@ -34,7 +34,6 @@ import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -808,25 +807,6 @@ public class ImapMailReceiverTests {
 		return folder;
 	}
 
-	@Test
-	public void testExecShutdown() {
-		ImapIdleChannelAdapter adapter = new ImapIdleChannelAdapter(new ImapMailReceiver());
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
-		adapter.setTaskScheduler(taskScheduler);
-		adapter.setReconnectDelay(1);
-		adapter.start();
-		ExecutorService exec = TestUtils.getPropertyValue(adapter, "sendingTaskExecutor", ExecutorService.class);
-		adapter.stop();
-		assertThat(exec.isShutdown()).isTrue();
-		adapter.start();
-		exec = TestUtils.getPropertyValue(adapter, "sendingTaskExecutor", ExecutorService.class);
-		adapter.stop();
-		assertThat(exec.isShutdown()).isTrue();
-
-		taskScheduler.shutdown();
-	}
-
 	@Test
 	public void testNullMessages() throws Exception {
 		Message message1 = mock(Message.class);