DSL: Mail - Add IMAP Idle Adapter

Polish IMAP test server and update it to support IDLE.

Polishing

Polishing - Fix IMAP Server

Changes required because receiver is now initialized.

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -94,8 +94,10 @@ public final class IntegrationFlowBuilder {
 	}
 
 	IntegrationFlowBuilder addComponents(Collection<Object> components) {
-		for (Object component : components) {
-			this.flow.addComponent(component);
+		if (components != null) {
+			for (Object component : components) {
+				this.flow.addComponent(component);
+			}
 		}
 		return this;
 	}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/ImapIdleChannelAdapterSpec.java

@@ -0,0 +1,124 @@
+/*
+ * 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.dsl.mail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+import javax.mail.Authenticator;
+import javax.mail.Session;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.integration.dsl.core.ComponentsRegistration;
+import org.springframework.integration.dsl.core.MessagingProducerSpec;
+import org.springframework.integration.dsl.support.PropertiesBuilder;
+import org.springframework.integration.dsl.support.PropertiesConfigurer;
+import org.springframework.integration.mail.ImapIdleChannelAdapter;
+import org.springframework.integration.mail.ImapMailReceiver;
+import org.springframework.integration.mail.SearchTermStrategy;
+import org.springframework.integration.transaction.TransactionSynchronizationFactory;
+
+/**
+ * @author Gary Russell
+ */
+public class ImapIdleChannelAdapterSpec
+		extends MessagingProducerSpec<ImapIdleChannelAdapterSpec, ImapIdleChannelAdapter>
+		implements ComponentsRegistration {
+
+	private final ImapMailReceiver receiver;
+
+	ImapIdleChannelAdapterSpec(ImapMailReceiver receiver) {
+		super(new ImapIdleChannelAdapter(receiver));
+		this.receiver = receiver;
+	}
+
+	public ImapIdleChannelAdapterSpec selectorExpression(String selectorExpression) {
+		this.receiver.setSelectorExpression(PARSER.parseExpression(selectorExpression));
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec session(Session session) {
+		this.receiver.setSession(session);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec javaMailProperties(Properties javaMailProperties) {
+		this.receiver.setJavaMailProperties(javaMailProperties);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec javaMailProperties(PropertiesConfigurer configurer) {
+		PropertiesBuilder properties = new PropertiesBuilder();
+		configurer.configure(properties);
+		return javaMailProperties(properties.get());
+	}
+
+	public ImapIdleChannelAdapterSpec javaMailAuthenticator(Authenticator javaMailAuthenticator) {
+		this.receiver.setJavaMailAuthenticator(javaMailAuthenticator);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec maxFetchSize(int maxFetchSize) {
+		this.receiver.setMaxFetchSize(maxFetchSize);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec shouldDeleteMessages(boolean shouldDeleteMessages) {
+		this.receiver.setShouldDeleteMessages(shouldDeleteMessages);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec searchTermStrategy(SearchTermStrategy searchTermStrategy) {
+		this.receiver.setSearchTermStrategy(searchTermStrategy);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec shouldMarkMessagesAsRead(boolean shouldMarkMessagesAsRead) {
+		this.receiver.setShouldMarkMessagesAsRead(shouldMarkMessagesAsRead);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec
+	transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) {
+		this.target.setTransactionSynchronizationFactory(transactionSynchronizationFactory);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec adviceChain(Advice... adviceChain) {
+		this.target.setAdviceChain(Arrays.asList(adviceChain));
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec sendingTaskExecutor(Executor sendingTaskExecutor) {
+		this.target.setSendingTaskExecutor(sendingTaskExecutor);
+		return this;
+	}
+
+	public ImapIdleChannelAdapterSpec shouldReconnectAutomatically(boolean shouldReconnectAutomatically) {
+		this.target.setShouldReconnectAutomatically(shouldReconnectAutomatically);
+		return this;
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.<Object>singletonList(this.receiver);
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.integration.dsl.mail;
 
+import org.springframework.integration.mail.ImapMailReceiver;
+
 /**
  * @author Gary Russell
  *
@@ -46,4 +48,16 @@ public class Mail {
 		return new ImapMailInboundChannelAdapterSpec(url);
 	}
 
+	public static ImapIdleChannelAdapterSpec imapIdleAdapter() {
+		return imapIdleAdapter(new ImapMailReceiver());
+	}
+
+	public static ImapIdleChannelAdapterSpec imapIdleAdapter(String url) {
+		return imapIdleAdapter(new ImapMailReceiver(url));
+	}
+
+	private static ImapIdleChannelAdapterSpec imapIdleAdapter(ImapMailReceiver imapMailReceiver) {
+		return new ImapIdleChannelAdapterSpec(imapMailReceiver);
+	}
+
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailInboundChannelAdapterSpec.java

@@ -15,11 +15,14 @@
  */
 package org.springframework.integration.dsl.mail;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Properties;
 
 import javax.mail.Authenticator;
 import javax.mail.Session;
 
+import org.springframework.integration.dsl.core.ComponentsRegistration;
 import org.springframework.integration.dsl.core.MessageSourceSpec;
 import org.springframework.integration.dsl.support.PropertiesBuilder;
 import org.springframework.integration.dsl.support.PropertiesConfigurer;
@@ -32,7 +35,8 @@ import org.springframework.integration.mail.MailReceivingMessageSource;
  */
 public abstract class MailInboundChannelAdapterSpec<S extends MailInboundChannelAdapterSpec<S, R>,
 		R extends AbstractMailReceiver>
-		extends MessageSourceSpec<S, MailReceivingMessageSource> {
+		extends MessageSourceSpec<S, MailReceivingMessageSource>
+		implements ComponentsRegistration {
 
 	protected volatile R receiver;
 
@@ -72,6 +76,11 @@ public abstract class MailInboundChannelAdapterSpec<S extends MailInboundChannel
 		return _this();
 	}
 
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.<Object>singletonList(this.receiver);
+	}
+
 	@Override
 	public MailReceivingMessageSource doGet() {
 		return new MailReceivingMessageSource(this.receiver);

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java

@@ -18,17 +18,22 @@ package org.springframework.integration.dsl.test.mail;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.endsWith;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Properties;
 
+import javax.mail.Flags;
 import javax.mail.internet.AddressException;
 import javax.mail.internet.InternetAddress;
 import javax.mail.internet.MimeMessage;
 import javax.mail.internet.MimeMessage.RecipientType;
+import javax.mail.search.AndTerm;
+import javax.mail.search.FlagTerm;
 import javax.mail.search.FromTerm;
+import javax.mail.search.SearchTerm;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -48,6 +53,7 @@ import org.springframework.integration.dsl.support.Pollers;
 import org.springframework.integration.dsl.test.mail.PoorMansMailServer.ImapServer;
 import org.springframework.integration.dsl.test.mail.PoorMansMailServer.Pop3Server;
 import org.springframework.integration.dsl.test.mail.PoorMansMailServer.SmtpServer;
+import org.springframework.integration.mail.ImapIdleChannelAdapter;
 import org.springframework.integration.mail.MailHeaders;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
@@ -71,15 +77,19 @@ public class MailTests {
 
 	private final static int smtpPort = SocketUtils.findAvailableTcpPort();
 
-	private static SmtpServer smtpServer = PoorMansMailServer.smtp(smtpPort);
+	private final static SmtpServer smtpServer = PoorMansMailServer.smtp(smtpPort);
 
 	private final static int pop3Port = SocketUtils.findAvailableTcpPort(smtpPort + 1);
 
-	private static Pop3Server pop3Server = PoorMansMailServer.pop3(pop3Port);
+	private final static Pop3Server pop3Server = PoorMansMailServer.pop3(pop3Port);
 
 	private final static int imapPort = SocketUtils.findAvailableTcpPort(pop3Port + 1);
 
-	private static ImapServer imapServer = PoorMansMailServer.imap(imapPort);
+	private final static ImapServer imapServer = PoorMansMailServer.imap(imapPort);
+
+	private final static int imapIdlePort = SocketUtils.findAvailableTcpPort(imapPort + 1);
+
+	private final static ImapServer imapIdleServer = PoorMansMailServer.imap(imapIdlePort);
 
 
 	@BeforeClass
@@ -96,6 +106,7 @@ public class MailTests {
 		smtpServer.stop();
 		pop3Server.stop();
 		imapServer.stop();
+		imapIdleServer.stop();
 	}
 
 	@Autowired
@@ -111,6 +122,12 @@ public class MailTests {
 	@Autowired
 	private PollableChannel imapChannel;
 
+	@Autowired
+	private PollableChannel imapIdleChannel;
+
+	@Autowired
+	private ImapIdleChannelAdapter imapIdleAdapter;
+
 	@Test
 	public void testOutbound() throws Exception {
 		assertEquals("localhost", TestUtils.getPropertyValue(this.sendMailHandler, "mailSender.host"));
@@ -162,6 +179,19 @@ public class MailTests {
 		assertEquals("foo\r\n", mm.getContent());
 	}
 
+	@Test
+	public void testImapIdle() throws Exception {
+		Message<?> message = this.imapIdleChannel.receive(10000);
+		assertNotNull(message);
+		MimeMessage mm = (MimeMessage) message.getPayload();
+		assertEquals("foo@bar", mm.getRecipients(RecipientType.TO)[0].toString());
+		assertEquals("bar@baz", mm.getFrom()[0].toString());
+		assertEquals("Test Email", mm.getSubject());
+		assertEquals("foo\r\n", mm.getContent());
+		this.imapIdleAdapter.stop();
+		assertFalse(TestUtils.getPropertyValue(this.imapIdleAdapter, "shouldReconnectAutomatically", Boolean.class));
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
@@ -193,7 +223,7 @@ public class MailTests {
 		public IntegrationFlow imapMailFlow() {
 			return IntegrationFlows
 					.from(Mail.imapInboundAdapter("imap://user:pw@localhost:" + imapPort + "/INBOX")
-									.searchTermStrategy((f,l) -> new FromTerm(fromAddress()))
+									.searchTermStrategy((f, l) -> fromAndNotSeenTerm())
 									.javaMailProperties(p -> p.put("mail.debug", "true")),
 							e -> e.autoStartup(true)
 									.poller(Pollers.fixedDelay(1000)))
@@ -201,13 +231,27 @@ public class MailTests {
 					.get();
 		}
 
-		private InternetAddress fromAddress() {
+		@Bean
+		public IntegrationFlow imapIdleFlow() {
+			return IntegrationFlows
+					.from(Mail.imapIdleAdapter("imap://user:pw@localhost:" + imapIdlePort + "/INBOX")
+							.searchTermStrategy((f, l) -> fromAndNotSeenTerm())
+							.javaMailProperties(p -> p.put("mail.debug", "true")
+									.put("mail.imap.connectionpoolsize", "5"))
+							.shouldReconnectAutomatically(false))
+					.channel(MessageChannels.queue("imapIdleChannel"))
+					.get();
+		}
+
+		private SearchTerm fromAndNotSeenTerm() {
 			try {
-				return new InternetAddress("bar@baz");
+				FromTerm fromTerm = new FromTerm(new InternetAddress("bar@baz"));
+				return new AndTerm(fromTerm, new FlagTerm(new Flags(Flags.Flag.SEEN), false));
 			}
 			catch (AddressException e) {
 				throw new RuntimeException(e);
 			}
+
 		}
 
 	}

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java

@@ -203,6 +203,8 @@ public class PoorMansMailServer {
 
 	public static class ImapServer extends MailServer {
 
+		private boolean seen;
+
 		public ImapServer(int port) throws IOException {
 			super(port);
 		}
@@ -222,35 +224,49 @@ public class PoorMansMailServer {
 			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);
-						System.out.println(line);
 						if (line.endsWith("CAPABILITY")) {
-							write("* CAPABILITY IMAP4rev1");
+							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("EXAMINE \"\"")) {
-							write("* 1 EXISTS");
-							write("* 1 RECENT");
-							write("* OK [UNSEEN 1]");
-							write(tag + "OK EXAMINE completed");
+						else if (line.endsWith("SELECT INBOX")) {
+							if (!seen) {
+								write("* 1 EXISTS");
+								write("* 1 RECENT");
+								write("* OK [UNSEEN 1]");
+							}
+							write(tag + "OK SELECT completed");
 						}
-						else if (line.endsWith("SEARCH FROM bar@baz ALL")) {
-							write("* SEARCH 1 1");
+						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,1")) {
+						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\" "
@@ -270,6 +286,7 @@ public class PoorMansMailServer {
 						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)");
@@ -287,6 +304,17 @@ public class PoorMansMailServer {
 						else if (line.contains("NOOP")) {
 							write(tag + "OK NOOP completed");
 						}
+						else if (line.endsWith("IDLE")) {
+							write("+ idling");
+							idleTag = tag;
+						}
+						else if (line.equals("DONE")) {
+							write(idleTag + "OK");
+						}
+						else if (line.contains("LOGOUT")) {
+							write(tag + "OK LOGOUT completed");
+							this.socket.close();
+						}
 					}
 				}
 				catch (IOException e) {