diff --git a/spring-integration-mail/.springBeans b/spring-integration-mail/.springBeans index 5a41740f54..56d56b57cd 100644 --- a/spring-integration-mail/.springBeans +++ b/spring-integration-mail/.springBeans @@ -9,6 +9,7 @@ src/test/java/org/springframework/integration/mail/config/MailOutboundWithJavamailProperties-context.xml src/test/java/org/springframework/integration/mail/config/MessageWithContentTypeTests-context.xml + src/test/java/org/springframework/integration/mail/config/incept5-config.xml diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java index b9b3bc691e..e7f09e1747 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java +++ b/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); + } + } } /** diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java index acf6f95392..55533a399b 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java +++ b/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); } } } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ResubmittingTask.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ResubmittingTask.java index e3a541dc4d..29dc7a25ba 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ResubmittingTask.java +++ b/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)); } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java index 2be8e63c20..2e743b09c4 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java +++ b/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()); + } }