From 93b1224e99ba021be2f1de363eb69d0394241eca Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 27 Aug 2014 19:16:13 +0300 Subject: [PATCH] INT-3501 Fix IMAP Idle JIRA: https://jira.spring.io/browse/INT-3501 There were several problems: When a `Folder` is open, an activity on the `Store` opens a new connection. The `PingTask` called `isConnected()` on the store, creating a new connection. The pings (NOOPs) were performed on this connection and therefore did NOT cancel the `IDLE`. In any case, the `IDLE` was issued on the folder not the store and the only way to cancel that IDLE is to cause its `waitIfIdle()` to be invoked. Conveniently, simply calling `isOpen()` invokes that method and cancels the IDLE. Finally, the `SimpleMessageCountListener` unnecessarily invoked `message.getLineCount()` when a simple `folder.isOpen()` is sufficient. Fixes: 1. Do not invoke `openSession` if `this.folder` is not null. 2. Change the `PingTask` to simply invoke `isOpen()`. 3. Move the `PingTask to the receiver for a more efficient algorithm instead of running on fixed interval. Rename it `IdleCanceler`. 4. Increase the PING timer from 10 to 120 seconds; add setters for it and the `reconnectionDelay`. TODO: Namespace support for 4.1. (Not to be back ported). Add a test IMAP server (ported from Java DSL and enhanced to simulate IDLE with a new message arriving for the first idle period. INT-3501 Polishing - PR Comments --- .../mail/AbstractMailReceiver.java | 2 +- .../mail/ImapIdleChannelAdapter.java | 35 +- .../integration/mail/ImapMailReceiver.java | 72 ++- .../mail/ImapMailReceiverTests.java | 135 +++++- .../integration/mail/PoorMansMailServer.java | 440 ++++++++++++++++++ 5 files changed, 643 insertions(+), 41 deletions(-) create mode 100644 spring-integration-mail/src/test/java/org/springframework/integration/mail/PoorMansMailServer.java 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 3af01359a1..66fe16c914 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 @@ -222,8 +222,8 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl } protected void openFolder() throws MessagingException { - this.openSession(); if (this.folder == null) { + openSession(); this.folder = obtainFolderInstance(); } if (this.folder == null || !this.folder.exists()) { 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 a9e367f230..73c1b90f21 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 @@ -26,7 +26,6 @@ import java.util.concurrent.ScheduledFuture; import javax.mail.FolderClosedException; import javax.mail.Message; import javax.mail.MessagingException; -import javax.mail.Store; import org.aopalliance.aop.Advice; @@ -63,6 +62,8 @@ import org.springframework.util.CollectionUtils; public class ImapIdleChannelAdapter extends MessageProducerSupport implements BeanClassLoaderAware, ApplicationEventPublisherAware { + private static final int DEFAULT_RECONNECT_DELAY = 10000; + private final IdleTask idleTask = new IdleTask(); private volatile Executor sendingTaskExecutor; @@ -77,14 +78,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be private final ImapMailReceiver mailReceiver; - private volatile int reconnectDelay = 10000; // milliseconds + private volatile long reconnectDelay = DEFAULT_RECONNECT_DELAY; // milliseconds private volatile ScheduledFuture receivingTask; - private volatile ScheduledFuture pingTask; - - private volatile long connectionPingInterval = 10000; - private final ExceptionAwarePeriodicTrigger receivingTaskTrigger = new ExceptionAwarePeriodicTrigger(); private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; @@ -127,6 +124,15 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be this.shouldReconnectAutomatically = shouldReconnectAutomatically; } + /** + * The time between connection attempts in milliseconds (default 10 seconds). + * @param reconnectDelay the reconnectDelay to set + * @since 3.0.5 + */ + public void setReconnectDelay(long reconnectDelay) { + this.reconnectDelay = reconnectDelay; + } + @Override public String getComponentType() { return "mail:imap-idle-channel-adapter"; @@ -154,14 +160,12 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be this.sendingTaskExecutor = Executors.newFixedThreadPool(1); } this.receivingTask = scheduler.schedule(new ReceivingTask(), this.receivingTaskTrigger); - this.pingTask = scheduler.scheduleAtFixedRate(new PingTask(), this.connectionPingInterval); } @Override // guarded by super#lifecycleLock protected void doStop() { this.receivingTask.cancel(true); - this.pingTask.cancel(true); try { this.mailReceiver.destroy(); } @@ -290,21 +294,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be } } - private class PingTask implements Runnable { - - @Override - public void run() { - try { - Store store = mailReceiver.getStore(); - if (store != null) { - store.isConnected(); - } - } - catch (Exception ignore) { - } - } - } - private class ExceptionAwarePeriodicTrigger implements Trigger { private volatile boolean delayNextExecution; diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java index c69f601482..a7c3904c6c 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java @@ -16,6 +16,9 @@ package org.springframework.integration.mail; +import java.util.Date; +import java.util.concurrent.ScheduledFuture; + import javax.mail.Flags; import javax.mail.Flags.Flag; import javax.mail.Folder; @@ -29,6 +32,8 @@ import javax.mail.search.FlagTerm; import javax.mail.search.NotTerm; import javax.mail.search.SearchTerm; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import com.sun.mail.imap.IMAPFolder; @@ -49,11 +54,21 @@ import com.sun.mail.imap.IMAPMessage; */ public class ImapMailReceiver extends AbstractMailReceiver { + private static final int DEFAULT_CANCEL_IDLE_INTERVAL = 120000; + + private final MessageCountListener messageCountListener = new SimpleMessageCountListener(); + + private final IdleCanceler idleCanceler = new IdleCanceler(); + private volatile boolean shouldMarkMessagesAsRead = true; private volatile SearchTermStrategy searchTermStrategy = new DefaultSearchTermStrategy(); - private final MessageCountListener messageCountListener = new SimpleMessageCountListener(); + private volatile long cancelIdleInterval = DEFAULT_CANCEL_IDLE_INTERVAL; + + private volatile TaskScheduler scheduler; + + private volatile ScheduledFuture pingTask; public ImapMailReceiver() { super(); @@ -101,6 +116,28 @@ public class ImapMailReceiver extends AbstractMailReceiver { this.shouldMarkMessagesAsRead = shouldMarkMessagesAsRead; } + /** + * IDLE commands will be terminated after this interval; useful in cases where a connection + * might be silently dropped. A new IDLE will usually immediately be processed. Specified + * in seconds; default 120 (2 minutes). RFC 2177 recommends an interval no larger than 29 minutes. + * @param cancelIdleInterval the cancelIdleInterval to set + * @since 3.0.5 + */ + public void setCancelIdleInterval(long cancelIdleInterval) { + this.cancelIdleInterval = cancelIdleInterval * 1000; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + this.scheduler = getTaskScheduler(); + if (this.scheduler == null) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + this.scheduler = scheduler; + } + } + /** * This method is unique to the IMAP receiver and only works if IMAP IDLE * is supported (see RFC 2177 for more detail). @@ -123,10 +160,15 @@ public class ImapMailReceiver extends AbstractMailReceiver { } imapFolder.addMessageCountListener(this.messageCountListener); try { + this.pingTask = this.scheduler.schedule(this.idleCanceler, + new Date(System.currentTimeMillis() + this.cancelIdleInterval)); imapFolder.idle(); } finally { imapFolder.removeMessageCountListener(this.messageCountListener); + if (this.pingTask != null) { + this.pingTask.cancel(true); + } } } @@ -168,23 +210,32 @@ public class ImapMailReceiver extends AbstractMailReceiver { } } + private class IdleCanceler implements Runnable { + @Override + public void run() { + try { + IMAPFolder folder = (IMAPFolder) getFolder(); + logger.debug("Canceling IDLE"); + if (folder != null) { + folder.isOpen(); // resets idle state + } + } + catch (Exception ignore) { + } + } + } /** * Callback used for handling the event-driven idle response. */ - private static class SimpleMessageCountListener extends MessageCountAdapter { + private class SimpleMessageCountListener extends MessageCountAdapter { @Override public void messagesAdded(MessageCountEvent event) { Message[] messages = event.getMessages(); - for (Message message : messages) { - try { - // this will return the flow to the idle call - message.getLineCount(); - } - catch (MessagingException e) { - // ignored; - } + if (messages.length > 0) { + // this will return the flow to the idle call + messages[0].getFolder().isOpen(); } } } @@ -255,5 +306,4 @@ public class ImapMailReceiver extends AbstractMailReceiver { } - } 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 66428f4aaa..de6833de2a 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 @@ -46,10 +46,19 @@ import javax.mail.MessagingException; import javax.mail.Multipart; import javax.mail.Store; import javax.mail.URLName; +import javax.mail.internet.AddressException; +import javax.mail.internet.InternetAddress; import javax.mail.internet.MimeMessage; +import javax.mail.search.AndTerm; +import javax.mail.search.FlagTerm; +import javax.mail.search.FromTerm; import javax.mail.search.SearchTerm; +import org.apache.commons.logging.Log; +import org.junit.AfterClass; +import org.junit.BeforeClass; import org.junit.Ignore; +import org.junit.Rule; import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; @@ -57,21 +66,25 @@ import org.mockito.stubbing.Answer; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; -import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.io.ClassPathResource; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.messaging.PollableChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.mail.ImapIdleChannelAdapter.ImapIdleExceptionEvent; +import org.springframework.integration.mail.PoorMansMailServer.ImapServer; import org.springframework.integration.mail.config.ImapIdleChannelAdapterParserTests; +import org.springframework.integration.test.support.LongRunningIntegrationTest; +import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.PollableChannel; +import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.FileCopyUtils; @@ -85,8 +98,72 @@ import com.sun.mail.imap.IMAPMessage; */ public class ImapMailReceiverTests { + @Rule + public final LongRunningIntegrationTest longRunningIntegrationTest = new LongRunningIntegrationTest(); + private final AtomicInteger failed = new AtomicInteger(0); + private final static int imapIdlePort = SocketUtils.findAvailableServerSocket(); + + private final static ImapServer imapIdleServer = PoorMansMailServer.imap(imapIdlePort); + + + @BeforeClass + public static void setup() throws InterruptedException { + int n = 0; + while (n++ < 100 && (!imapIdleServer.isListening())) { + Thread.sleep(100); + } + assertTrue(n < 100); + } + + @AfterClass + public static void tearDown() { + imapIdleServer.stop(); + } + + @Test + public void testIdleWithServer() throws Exception { + Properties mailProps = new Properties(); + mailProps.put("mail.debug", "true"); + mailProps.put("mail.imap.connectionpool.debug", "true"); + ImapMailReceiver receiver = new ImapMailReceiver("imap://user:pw@localhost:" + imapIdlePort + "/INBOX"); + receiver.setJavaMailProperties(mailProps); + receiver.setMaxFetchSize(1); + receiver.setShouldDeleteMessages(false); + receiver.setShouldMarkMessagesAsRead(true); + receiver.setSearchTermStrategy(new SearchTermStrategy() { + + @Override + public SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder) { + try { + FromTerm fromTerm = new FromTerm(new InternetAddress("bar@baz")); + AndTerm andTerm = new AndTerm(fromTerm, new FlagTerm(new Flags(Flags.Flag.SEEN), false)); + return andTerm; + } + catch (AddressException e) { + throw new RuntimeException(e); + } + } + }); + receiver.setCancelIdleInterval(8); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + setUpScheduler(receiver, taskScheduler); + receiver.afterPropertiesSet(); + Log logger = spy(TestUtils.getPropertyValue(receiver, "logger", Log.class)); + new DirectFieldAccessor(receiver).setPropertyValue("logger", logger); + ImapIdleChannelAdapter adapter = new ImapIdleChannelAdapter(receiver); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + assertNotNull(channel.receive(6000)); + assertNotNull(channel.receive(6000)); // new message after idle + assertNull(channel.receive(10000)); // no new message after second and third idle + verify(logger).debug("Canceling IDLE"); + taskScheduler.shutdown(); + } + @Test public void receiveAndMarkAsReadDontDelete() throws Exception{ AbstractMailReceiver receiver = new ImapMailReceiver(); @@ -113,6 +190,7 @@ public class ImapMailReceiverTests { final Message[] messages = new Message[]{msg1, msg2}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock()); int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode"); @@ -125,12 +203,14 @@ public class ImapMailReceiverTests { }).when(receiver).openFolder(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -187,6 +267,7 @@ public class ImapMailReceiverTests { Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock()); int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode"); @@ -198,12 +279,14 @@ public class ImapMailReceiverTests { }).when(receiver).openFolder(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -233,18 +316,21 @@ public class ImapMailReceiverTests { Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } }).when(receiver).openFolder(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -273,6 +359,7 @@ public class ImapMailReceiverTests { Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock()); int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode"); @@ -284,12 +371,14 @@ public class ImapMailReceiverTests { }).when(receiver).openFolder(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -318,6 +407,7 @@ public class ImapMailReceiverTests { Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock()); int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode"); @@ -329,12 +419,14 @@ public class ImapMailReceiverTests { }).when(receiver).openFolder(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -347,7 +439,7 @@ public class ImapMailReceiverTests { @Test @Ignore public void testMessageHistory() throws Exception{ - ApplicationContext context = + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class); ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class); @@ -365,6 +457,7 @@ public class ImapMailReceiverTests { final Message[] messages = new Message[]{mailMessage}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accesor = new DirectFieldAccessor((invocation.getMock())); IMAPFolder folder = mock(IMAPFolder.class); @@ -375,12 +468,14 @@ public class ImapMailReceiverTests { }).when(receiver).openFolder(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -396,11 +491,12 @@ public class ImapMailReceiverTests { assertNotNull(componentHistoryRecord); assertEquals("mail:imap-idle-channel-adapter", componentHistoryRecord.get("type")); adapter.stop(); + context.close(); } @Test public void testIdleChannelAdapterException() throws Exception{ - ApplicationContext context = + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class); ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class); @@ -431,12 +527,14 @@ public class ImapMailReceiverTests { folderField.set(receiver, folder); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return true; } }).when(folder).isOpen(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -451,12 +549,14 @@ public class ImapMailReceiverTests { final Message[] messages = new Message[]{mailMessage}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -467,11 +567,12 @@ public class ImapMailReceiverTests { assertNotNull(replMessage); assertEquals("Failed", ((Exception) replMessage.getPayload()).getCause().getMessage()); adapter.stop(); + context.close(); } @Test public void testNoInitialIdleDelayWhenRecentNotSupported() throws Exception{ - ApplicationContext context = + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class); ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class); @@ -499,6 +600,7 @@ public class ImapMailReceiverTests { storeField.set(receiver, store); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return folder; } @@ -511,6 +613,7 @@ public class ImapMailReceiverTests { final AtomicInteger shouldFindMessagesCounter = new AtomicInteger(2); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { /* * Return the message from first invocation of waitForMessages() @@ -528,12 +631,14 @@ public class ImapMailReceiverTests { }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } }).when(receiver).fetchMessages(messages); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { Thread.sleep(5000); shouldFindMessagesCounter.set(1); @@ -552,11 +657,12 @@ public class ImapMailReceiverTests { assertNull(channel.receive(3000)); assertNotNull(channel.receive(6000)); adapter.stop(); + context.close(); } @Test public void testInitialIdleDelayWhenRecentIsSupported() throws Exception{ - ApplicationContext context = + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class); ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class); @@ -584,6 +690,7 @@ public class ImapMailReceiverTests { storeField.set(receiver, store); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return folder; } @@ -595,12 +702,14 @@ public class ImapMailReceiverTests { final Message[] messages = new Message[]{mailMessage}; doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } @@ -608,6 +717,7 @@ public class ImapMailReceiverTests { final CountDownLatch idles = new CountDownLatch(2); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { idles.countDown(); Thread.sleep(5000); @@ -625,6 +735,7 @@ public class ImapMailReceiverTests { assertNotNull(channel.receive(5000)); assertTrue(idles.await(5, TimeUnit.SECONDS)); adapter.stop(); + context.close(); } @Test @@ -670,6 +781,7 @@ public class ImapMailReceiverTests { receiver.afterPropertiesSet(); new Thread(new Runnable() { + @Override public void run(){ try { receiver.receive(); @@ -684,6 +796,7 @@ public class ImapMailReceiverTests { }).start(); new Thread(new Runnable() { + @Override public void run(){ try { receiver.destroy(); @@ -716,6 +829,7 @@ public class ImapMailReceiverTests { doAnswer(new Answer () { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { OutputStream os = (OutputStream) invocation.getArguments()[0]; FileCopyUtils.copy(new ClassPathResource("test.mail").getInputStream(), os); @@ -745,4 +859,13 @@ public class ImapMailReceiverTests { assertTrue(exec.isShutdown()); } + private void setUpScheduler(ImapMailReceiver mailReceiver, ThreadPoolTaskScheduler taskScheduler) { + taskScheduler.setPoolSize(5); + taskScheduler.initialize(); + BeanFactory bf = mock(BeanFactory.class); + when(bf.containsBean("taskScheduler")).thenReturn(true); + when(bf.getBean("taskScheduler", TaskScheduler.class)).thenReturn(taskScheduler); + mailReceiver.setBeanFactory(bf); + } + } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/PoorMansMailServer.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/PoorMansMailServer.java new file mode 100644 index 0000000000..5c309844ce --- /dev/null +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/PoorMansMailServer.java @@ -0,0 +1,440 @@ +/* + * Copyright 2014 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mail; + +import java.io.BufferedReader; +import java.io.BufferedWriter; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStreamWriter; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import javax.net.ServerSocketFactory; + +import org.springframework.util.Base64Utils; + +/** + * @author Gary Russell + * + */ +public class PoorMansMailServer { + + public static SmtpServer smtp(int port) { + try { + return new SmtpServer(port); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public static Pop3Server pop3(int port) { + try { + return new Pop3Server(port); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public static ImapServer imap(int port) { + try { + return new ImapServer(port); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public static class SmtpServer extends MailServer { + + public SmtpServer(int port) throws IOException { + super(port); + } + + @Override + protected MailHandler mailHandler(Socket socket) { + return new SmtpHandler(socket); + } + + public class SmtpHandler extends MailHandler { + + public SmtpHandler(Socket socket) { + super(socket); + } + + @Override + void doRun() { + try { + write("220 foo SMTP"); + while (!socket.isClosed()) { + String line = reader.readLine(); + if (line.contains("EHLO")) { + write("250-foo hello [0,0,0,0], foo"); + write("250-AUTH LOGIN PLAIN"); + write("250 OK"); + } + else if (line.contains("MAIL FROM")) { + write("250 OK"); + } + else if (line.contains("RCPT TO")) { + write("250 OK"); + } + else if (line.contains("AUTH LOGIN")) { + write("334 VXNlcm5hbWU6"); + } + else if (line.contains("dXNlcg==")) { // base64 'user' + sb.append("user:"); + sb.append((new String(Base64Utils.decode(line.getBytes())))); + sb.append("\n"); + write("334 UGFzc3dvcmQ6"); + } + else if (line.contains("cHc=")) { // base64 'pw' + sb.append("password:"); + sb.append((new String(Base64Utils.decode(line.getBytes())))); + sb.append("\n"); + write("235"); + } + else if (line.equals("DATA")) { + write("354"); + } + else if (line.equals(".")) { + write("250"); + } + else if (line.equals("QUIT")) { + write("221"); + socket.close(); + } + else { + sb.append(line); + sb.append("\n"); + } + } + messages.add(sb.toString()); + } + catch (IOException e) { + e.printStackTrace(); + } + } + + } + + } + + public static class Pop3Server extends MailServer { + + public Pop3Server(int port) throws IOException { + super(port); + } + + @Override + protected MailHandler mailHandler(Socket socket) { + return new Pop3Handler(socket); + } + + public class Pop3Handler extends MailHandler { + + public Pop3Handler(Socket socket) { + super(socket); + } + + @Override + void doRun() { + try { + write("+OK POP3"); + while (!socket.isClosed()) { + String line = reader.readLine(); + if ("CAPA".equals(line)) { + write("+OK"); + write("USER"); + write("."); + } + else if ("USER user".equals(line)) { + write("+OK"); + } + else if ("PASS pw".equals(line)) { + write("+OK"); + } + else if ("STAT".equals(line)) { + write("+OK 1 3"); + } + else if ("NOOP".equals(line)) { + write("+OK"); + } + else if ("RETR 1".equals(line)) { + write("+OK"); + write(MESSAGE); + write("."); + } + else if ("QUIT".equals(line)) { + write("+OK"); + socket.close(); + } + } + } + catch (IOException e) { + e.printStackTrace(); + } + } + + } + + } + + public static class ImapServer extends MailServer { + + private boolean seen; + + private boolean idled; + + public ImapServer(int port) throws IOException { + super(port); + } + + @Override + protected MailHandler mailHandler(Socket socket) { + return new ImapHandler(socket); + } + + public class ImapHandler extends MailHandler { + + public ImapHandler(Socket socket) { + super(socket); + } + + @Override + void doRun() { + try { + write("* OK IMAP4rev1 Service Ready"); + String idleTag = ""; + while (!socket.isClosed()) { + String line = reader.readLine(); + if (line == null) { + break; + } + String tag = line.substring(0, line.indexOf(" ") + 1); + if (line.endsWith("CAPABILITY")) { + write("* CAPABILITY IDLE IMAP4rev1"); + write(tag + "OK CAPABILITY completed"); + } + else if (line.endsWith("LOGIN user pw")) { + write(tag + "OK LOGIN completed"); + } + else if (line.endsWith("LIST \"\" INBOX")) { + write("* LIST \"/\" \"INBOX\""); + write(tag + "OK LIST completed"); + } + else if (line.endsWith("LIST \"\" \"\"")) { + write("* LIST \"/\" \"\""); + write(tag + "OK LIST completed"); + } + else if (line.endsWith("SELECT INBOX")) { + write("* 1 EXISTS"); + if (!seen) { + write("* 1 RECENT"); + write("* OK [UNSEEN 1]"); + } + else { + write("* OK"); + } + write(tag + "OK SELECT completed"); + } + else if (line.endsWith("EXAMINE INBOX")) { + write(tag + "OK"); + } + else if (line.endsWith("SEARCH FROM bar@baz UNSEEN ALL")) { + if (seen) { + write("* SEARCH"); + } + else { + write("* SEARCH 1"); + } + write(tag + "OK SEARCH completed"); + } + else if (line.contains("FETCH 1 (ENVELOPE")) { + write("* 1 FETCH (RFC822.SIZE 6909 INTERNALDATE \"27-May-2013 09:45:41 +0000\" " + + "FLAGS (\\Seen) " + + "ENVELOPE (\"Mon, 27 May 2013 15:14:49 +0530\" " + + "\"Test Email\" ((\"Foo\" NIL \"foo\" \"bar.tv\")) " + + "((\"Foo\" NIL \"foo\" \"bar.tv\")) " + + "((\"Foo\" NIL \"foo\" \"bar.tv\")) " + + "((\"Bar\" NIL \"bar\" \"baz.net\")) NIL NIL " + + "\"<4DA0A7E4.3010506@baz.net>\" " + + "\"\") " + + "BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL \"7BIT\" 1176 43)))"); + write(tag + "OK FETCH completed"); + } + else if (line.contains("FETCH 2 (BODYSTRUCTURE)")) { + write("* 2 FETCH " + + "BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL \"7BIT\" 1176 43)))"); + write(tag + "OK FETCH completed"); + } + else if (line.contains("STORE 1 +FLAGS (\\Flagged)")) { + write("* 1 FETCH (FLAGS (\\Flagged))"); + write(tag + "OK STORE completed"); + } + else if (line.contains("STORE 1 +FLAGS (\\Seen)")) { + write("* 1 FETCH (FLAGS (\\Flagged \\Seen))"); + write(tag + "OK STORE completed"); + seen = true; + } + else if (line.contains("FETCH 1 FLAGS")) { + write("* 1 FLAGS(\\Seen)"); + write(tag + "OK FETCH completed"); + } + else if (line.contains("FETCH 1 (BODY.PEEK")) { + write("* 1 FETCH (BODY[]<0> {" + (MESSAGE.length() + 2) + "}"); + write(MESSAGE); + write(")"); + write(tag + "OK FETCH completed"); + } + else if (line.contains("CLOSE")) { + write(tag + "OK CLOSE completed"); + } + else if (line.contains("NOOP")) { + write(tag + "OK NOOP completed"); + } + else if (line.endsWith("IDLE")) { + write("+ idling"); + idleTag = tag; + if (!idled) { + try { + Thread.sleep(3000); + write("* 2 EXISTS"); + seen = false; + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + idled = true; + } + else if (line.equals("DONE")) { + write(idleTag + "OK"); + } + else if (line.contains("LOGOUT")) { + write(tag + "OK LOGOUT completed"); + this.socket.close(); + } + } + } + catch (IOException e) { + e.printStackTrace(); + } + } + + } + + } + + public abstract static class MailServer implements Runnable { + + private final ServerSocket socket; + + private final ExecutorService exec = Executors.newCachedThreadPool(); + + protected final List messages = new ArrayList(); + + private volatile boolean listening; + + public MailServer(int port) throws IOException { + this.socket = ServerSocketFactory.getDefault().createServerSocket(port); + this.listening = true; + exec.execute(this); + } + + public boolean isListening() { + return listening; + } + + public List getMessages() { + return messages; + } + + @Override + public void run() { + try { + while (!socket.isClosed()) { + Socket socket = this.socket.accept(); + exec.execute(mailHandler(socket)); + } + } + catch (IOException e) { + this.listening = false; + } + } + + protected abstract MailHandler mailHandler(Socket socket); + + public void stop() { + try { + this.socket.close(); + } + catch (IOException e) { + e.printStackTrace(); + } + this.exec.shutdownNow(); + } + + public abstract class MailHandler implements Runnable { + + protected static final String MESSAGE = "To: foo@bar\r\nFrom: bar@baz\r\nSubject: Test Email\r\n\r\nfoo"; + + protected final Socket socket; + + private BufferedWriter writer; + + protected StringBuilder sb = new StringBuilder(); + + protected BufferedReader reader; + + public MailHandler(Socket socket) { + this.socket = socket; + } + + @Override + public void run() { + try { + this.reader = new BufferedReader(new InputStreamReader(this.socket.getInputStream())); + this.writer = new BufferedWriter(new OutputStreamWriter(this.socket.getOutputStream())); + } + catch (IOException e) { + e.printStackTrace(); + } + doRun(); + } + + protected void write(String str) throws IOException { + this.writer.write(str); + this.writer.write("\r\n"); + this.writer.flush(); + } + + abstract void doRun(); + + } + + } + + private PoorMansMailServer() { + } + +}