From e1b51bec3afccd9f7fda9b21be79165ea7c9b6fd Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 1 Mar 2011 09:02:58 -0500 Subject: [PATCH] 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 | 1 + .../mail/AbstractMailReceiver.java | 70 ++++++++++--------- .../mail/ImapIdleChannelAdapter.java | 8 +-- .../integration/mail/ResubmittingTask.java | 8 ++- .../mail/ImapMailReceiverTests.java | 50 +++++++++++++ 5 files changed, 97 insertions(+), 40 deletions(-) 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()); + } }