diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 6b1a95e..f2f4c2d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -94,8 +94,10 @@ public final class IntegrationFlowBuilder { } IntegrationFlowBuilder addComponents(Collection components) { - for (Object component : components) { - this.flow.addComponent(component); + if (components != null) { + for (Object component : components) { + this.flow.addComponent(component); + } } return this; } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/ImapIdleChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/ImapIdleChannelAdapterSpec.java new file mode 100644 index 0000000..5453c6c --- /dev/null +++ b/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 + 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 getComponentsToRegister() { + return Collections.singletonList(this.receiver); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java index 238e385..21e5c9d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java +++ b/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); + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailInboundChannelAdapterSpec.java index 6099cda..d2defb0 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailInboundChannelAdapterSpec.java +++ b/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, R extends AbstractMailReceiver> - extends MessageSourceSpec { + extends MessageSourceSpec + implements ComponentsRegistration { protected volatile R receiver; @@ -72,6 +76,11 @@ public abstract class MailInboundChannelAdapterSpec getComponentsToRegister() { + return Collections.singletonList(this.receiver); + } + @Override public MailReceivingMessageSource doGet() { return new MailReceivingMessageSource(this.receiver); diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java index 05d4433..3225fc4 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java +++ b/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); } + } } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java index 9caaf07..7dbfd7b 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java +++ b/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) {