INT-1801 fixed the NPE during shutdown of Mail receiver, fixed the dead code in ImapIdleChannelAdapter, polished IdleTask (it now throws IllegalStateException which means resubmit after delay or SI MessagingException if resubmit flag is set to 'false')

spring-integration-mail/.springBeans

@@ -9,6 +9,7 @@
 	<configs>
 		<config>src/test/java/org/springframework/integration/mail/config/MailOutboundWithJavamailProperties-context.xml</config>
 		<config>src/test/java/org/springframework/integration/mail/config/MessageWithContentTypeTests-context.xml</config>
+		<config>src/test/java/org/springframework/integration/mail/config/incept5-config.xml</config>
 	</configs>
 	<configSets>
 	</configSets>

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

@@ -207,42 +207,46 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		this.folder.open(this.folderOpenMode);
 	}
 	
-	public synchronized Message[] receive() {
-		try {
-			this.openFolder();
-			if (logger.isInfoEnabled()) {
-				logger.info("attempting to receive mail from folder [" + this.getFolder().getFullName() + "]");
-			}
-			Message[] messages = this.searchForNewMessages();
-			if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
-				Message[] reducedMessages = new Message[this.maxFetchSize];
-				System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
-				messages = reducedMessages;
-			}
-			if (logger.isDebugEnabled()) {
-				logger.debug("found " + messages.length + " new messages");
-			}
-			if (messages.length > 0) {
-				this.fetchMessages(messages);
-			}
+	public synchronized Message[] receive() {	
+		synchronized (this.initializationMonitor) {
+			try {
+				this.openFolder();
+				if (logger.isInfoEnabled()) {
+					logger.info("attempting to receive mail from folder [" + this.getFolder().getFullName() + "]");
+				}
+				Message[] messages = this.searchForNewMessages();
+				
 
-			Message[] copiedMessages = new Message[messages.length];
-			for (int i = 0; i < messages.length; i++) {
-				this.setAdditionalFlags(messages[i]);
-				copiedMessages[i] = new MimeMessage((MimeMessage) messages[i]);
+				if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
+					Message[] reducedMessages = new Message[this.maxFetchSize];
+					System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
+					messages = reducedMessages;
+				}
+				if (logger.isDebugEnabled()) {
+					logger.debug("found " + messages.length + " new messages");
+				}
+				if (messages.length > 0) {
+					this.fetchMessages(messages);
+				}
+
+				Message[] copiedMessages = new Message[messages.length];
+				for (int i = 0; i < messages.length; i++) {
+					this.setAdditionalFlags(messages[i]);
+					copiedMessages[i] = new MimeMessage((MimeMessage) messages[i]);
+				}
+				if (this.shouldDeleteMessages()) {
+					this.deleteMessages(messages);
+				}
+				return copiedMessages;
 			}
-			if (this.shouldDeleteMessages()) {
-				this.deleteMessages(messages);
+			catch (Exception e) {
+				throw new org.springframework.integration.MessagingException(
+						"failure occurred while receiving from folder", e);
 			}
-			return copiedMessages;
-		}
-		catch (Exception e) {
-			throw new org.springframework.integration.MessagingException(
-					"failure occurred while receiving from folder", e);
-		}
-		finally {
-			MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
-		}
+			finally {
+				MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
+			}
+		}	
 	}
 
 	/**

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

@@ -16,10 +16,8 @@
 
 package org.springframework.integration.mail;
 
-import java.security.ProviderException;
 import java.util.Date;
 import java.util.concurrent.Executor;
-import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledFuture;
 
 import javax.mail.FolderClosedException;
@@ -94,7 +92,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 		Assert.notNull(scheduler, "'taskScheduler' must not be null" );
 		ResubmittingTask task = new ResubmittingTask(this.idleTask, scheduler, reconnectDelay);
 		task.setTaskExecutor(taskExecutor);
-		scheduledFuture = scheduler.schedule(new ResubmittingTask(this.idleTask, scheduler, reconnectDelay), new Date());
+		scheduledFuture = scheduler.schedule(task, new Date());
 	}
 
 	@Override // guarded by super#lifecycleLock
@@ -121,10 +119,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport {
 			} catch (MessagingException e) {
 				ImapIdleChannelAdapter.this.handleMailMessagingException(e);
 				if (shouldReconnectAutomatically){
-					throw new ProviderException("Failure in 'idle' task. Will resubmit", e);
+					throw new IllegalStateException("Failure in 'idle' task. Will resubmit", e);
 				}
 				else {
-					throw new RejectedExecutionException("Failure in 'idle' task. Will NOT resubmit", e);
+					throw new org.springframework.integration.MessagingException("Failure in 'idle' task. Will NOT resubmit", e);
 				}
 			}
 		}

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

@@ -15,7 +15,6 @@
  */
 package org.springframework.integration.mail;
 
-import java.security.ProviderException;
 import java.util.Date;
 import java.util.concurrent.Executor;
 
@@ -30,6 +29,11 @@ import org.springframework.scheduling.TaskScheduler;
  * @author Oleg Zhurakousky
  * @since 2.0.2
  *
+ * Will run the provided task right away resubmitting it after each successful execution 
+ * or after receiving a {@link IllegalStateException}. If resubmission is after exception then 
+ * the delay will be applied before resubmission. This is useful when the underlying task deals 
+ * with reconnection logic 
+ * Currently only used to manage IDLE task of ImapIdleChannelAdapter
  */
 class ResubmittingTask implements Runnable {
 	private static final Log logger = LogFactory.getLog(ResubmittingTask.class);
@@ -62,7 +66,7 @@ class ResubmittingTask implements Runnable {
 			logger.debug("Task completed successfully. Re-scheduling it again right away");
 			scheduler.schedule(this, new Date());
 		}
-		catch (ProviderException e) { //run again after a delay
+		catch (IllegalStateException e) { //run again after a delay
 			logger.warn("Failed to execute IDLE task. Will atempt to resubmit in " + delay + " milliseconds", e);
 			scheduler.schedule(this, new Date(System.currentTimeMillis() + delay));
 		}

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

@@ -25,20 +25,26 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.mail.Flags;
 import javax.mail.Flags.Flag;
 import javax.mail.Folder;
 import javax.mail.Message;
+import javax.mail.Store;
+import javax.mail.URLName;
 import javax.mail.internet.MimeMessage;
+import javax.mail.search.SearchTerm;
 
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessagingException;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PollableChannel;
@@ -56,6 +62,8 @@ import com.sun.mail.imap.IMAPFolder;
  */
 public class ImapMailReceiverTests {
 	
+	private AtomicInteger failed = new AtomicInteger(0);
+	
 	@Test
 	public void receiveAndMarkAsReadDontDelete() throws Exception{
 		AbstractMailReceiver receiver = new ImapMailReceiver();
@@ -344,4 +352,46 @@ public class ImapMailReceiverTests {
 		assertNotNull(replMessage);
 		assertEquals("Failed", ((Exception) replMessage.getPayload()).getCause().getMessage());
 	}
+	
+	@Test // see INT-1801
+	public void testImapLifecycleForRaceCondition() throws Exception{
+	
+		for (int i = 0; i < 1000; i++) {
+			final ImapMailReceiver receiver = new ImapMailReceiver("imap://foo");
+			Store store = mock(Store.class);
+			Folder folder = mock(Folder.class);
+			when(folder.exists()).thenReturn(true);
+			when(folder.isOpen()).thenReturn(true);
+			when(folder.search((SearchTerm) Mockito.any())).thenReturn(new Message[]{});
+			when(store.getFolder(Mockito.any(URLName.class))).thenReturn(folder);
+			DirectFieldAccessor df = new DirectFieldAccessor(receiver);
+			df.setPropertyValue("store", store);
+			
+			new Thread(new Runnable() {
+				public void run(){
+					try {
+						receiver.receive();
+					} catch (MessagingException e) {
+						if (e.getCause() instanceof NullPointerException){
+							e.printStackTrace();
+							failed.getAndIncrement();
+						}
+					}
+					
+				}
+			}).start();
+			
+			new Thread(new Runnable() {
+				public void run(){
+					try {
+						receiver.destroy();
+					} catch (Exception ignore) {
+						// ignore
+						ignore.printStackTrace();
+					}
+				}
+			}).start();
+		}
+		assertEquals(0, failed.get());
+	}
 }