diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/DefaultFolderConnection.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/DefaultFolderConnection.java new file mode 100644 index 0000000000..2eaa44ae85 --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/DefaultFolderConnection.java @@ -0,0 +1,155 @@ +/* + * Copyright 2002-2007 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.adapter.mail; + +import java.util.Properties; + +import javax.mail.Folder; +import javax.mail.Message; +import javax.mail.MessagingException; +import javax.mail.Session; +import javax.mail.Store; +import javax.mail.URLName; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.adapter.mail.monitor.AsyncMonitoringStrategy; +import org.springframework.integration.adapter.mail.monitor.MailTransportUtils; +import org.springframework.integration.adapter.mail.monitor.MonitoringStrategy; +import org.springframework.util.Assert; + +/** + * @author Jonas Partner + * + */ +public class DefaultFolderConnection implements Lifecycle, InitializingBean, + DisposableBean, FolderConnection { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final URLName storeUri; + + private final Session session; + + private final MonitoringStrategy monitoringStrategy; + + private final boolean polling; + + private Store store; + + private Folder folder; + + public DefaultFolderConnection(String storeUri, Properties javaMailProperties, + MonitoringStrategy monitoringStrategy, boolean polling) { + this.storeUri = new URLName(storeUri); + this.session = Session.getInstance(javaMailProperties); + this.monitoringStrategy = monitoringStrategy; + this.polling = polling; + if (!polling + && monitoringStrategy.getClass().isAssignableFrom(AsyncMonitoringStrategy.class)) { + throw new ConfigurationException( + "Folder connection requires an AsyncMonitoringStragey if polling is disabled"); + } + } + + public void afterPropertiesSet() throws Exception { + + Assert.notNull(storeUri, "Property 'storeUri' is required"); + Assert.notNull(session, "Property 'JavaMailProperties' is required"); + Assert.notNull(monitoringStrategy, + "An instantce of MonitoringStrategy' is required"); + start(); + } + + /* (non-Javadoc) + * @see org.springframework.integration.adapter.mail.FolderConnectionI#receive() + */ + public synchronized Message[] receive() { + if (!isRunning()) { + throw new org.springframework.integration.message.MessagingException( + "Folder connection is not running"); + } + + try { + if (!polling) { + ((AsyncMonitoringStrategy) monitoringStrategy) + .waitForNewMessages(folder); + } + return monitoringStrategy.receive(folder); + } catch (Exception e) { + throw new org.springframework.integration.message.MessagingException( + "Exception receiving from folder", e); + } + + } + + public void destroy() throws Exception { + stop(); + } + + public synchronized boolean isRunning() { + return (folder!= null && folder.isOpen()); + } + + public synchronized void start() { + try { + openSession(); + openFolder(); + } catch (MessagingException messageE) { + throw new org.springframework.integration.message.MessagingException( + "Excpetion starting MailSource", messageE); + } + + } + + public synchronized void stop() { + MailTransportUtils.closeFolder(folder); + MailTransportUtils.closeService(store); + folder = null; + store = null; + } + + private void openFolder() throws MessagingException { + if (folder != null && folder.isOpen()) { + return; + } + folder = store.getFolder(storeUri); + if (folder == null || !folder.exists()) { + throw new IllegalStateException("No default folder to receive from"); + } + if (logger.isDebugEnabled()) { + logger.debug("Opening folder [" + + MailTransportUtils.toPasswordProtectedString(storeUri) + + "]"); + } + folder.open(monitoringStrategy.getFolderOpenMode()); + } + + private void openSession() throws MessagingException { + store = session.getStore(storeUri); + if (logger.isDebugEnabled()) { + logger.debug("Connecting to store [" + + MailTransportUtils.toPasswordProtectedString(storeUri) + + "]"); + } + store.connect(); + } + +} diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/FolderConnection.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/FolderConnection.java new file mode 100644 index 0000000000..c07d48f78d --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/FolderConnection.java @@ -0,0 +1,32 @@ +/* + * Copyright 2002-2007 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.adapter.mail; + +import javax.mail.Folder; +import javax.mail.Message; + +import org.springframework.context.Lifecycle; + +/** + * * Encapsulates state for a restartable connection to a {@link Folder} and ensures thread safety + * @author Jonas Partner + * + */ +public interface FolderConnection extends Lifecycle{ + + Message[] receive(); + +} \ No newline at end of file diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/PollingMailSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/PollingMailSource.java index c82566175b..0fa5f86466 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/PollingMailSource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/PollingMailSource.java @@ -16,27 +16,16 @@ package org.springframework.integration.adapter.mail; -import java.util.Properties; - -import javax.mail.Folder; -import javax.mail.MessagingException; -import javax.mail.Session; -import javax.mail.Store; -import javax.mail.URLName; import javax.mail.internet.MimeMessage; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.DisposableBean; -import org.springframework.context.Lifecycle; import org.springframework.integration.adapter.mail.monitor.DefaultLocalMailMessageStore; import org.springframework.integration.adapter.mail.monitor.LocalMailMessageStore; -import org.springframework.integration.adapter.mail.monitor.MailTransportUtils; import org.springframework.integration.adapter.mail.monitor.MonitoringStrategy; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.PollableSource; -import org.springframework.util.Assert; /** * {@link MessageSource} implementation which delegates to a @@ -47,26 +36,18 @@ import org.springframework.util.Assert; * @author Jonas Partner */ @SuppressWarnings("unchecked") -public class PollingMailSource implements PollableSource, DisposableBean, Lifecycle { +public class PollingMailSource implements PollableSource { private final Log logger = LogFactory.getLog(this.getClass()); - private final MonitoringStrategy monitoringStrategy; - - private Session session; - - private Store store; - - private Folder folder; - - private URLName storeUri; + private final FolderConnection folderConnection; private MailMessageConverter converter = new DefaultMailMessageConverter(); private LocalMailMessageStore mailMessageStore = new DefaultLocalMailMessageStore(); - public PollingMailSource(MonitoringStrategy monitoringStrategy) { - this.monitoringStrategy = monitoringStrategy; + public PollingMailSource(FolderConnection folderConnetion) { + this.folderConnection = folderConnetion; } @SuppressWarnings("unchecked") @@ -75,12 +56,12 @@ public class PollingMailSource implements PollableSource, DisposableBean, Lifecy javax.mail.Message mailMessage = mailMessageStore.getNext(); if (mailMessage == null) { try { - javax.mail.Message[] messages = monitoringStrategy.receive(folder); + javax.mail.Message[] messages = folderConnection.receive(); mailMessageStore.addLast(messages); mailMessage = mailMessageStore.getNext(); - } - catch (Exception e) { - throw new org.springframework.integration.message.MessagingException("Excpetion receiving mail", e); + } catch (Exception e) { + throw new org.springframework.integration.message.MessagingException( + "Excpetion receiving mail", e); } } if (mailMessage != null) { @@ -92,77 +73,10 @@ public class PollingMailSource implements PollableSource, DisposableBean, Lifecy return received; } - public void setJavaMailProperties(Properties javaMailProperties) { - session = Session.getInstance(javaMailProperties, null); - } - - public void setJavaMailsession(Session session) { - this.session = session; - } - - public void setStoreUri(String storeUri) { - this.storeUri = new URLName(storeUri); - } - - public void afterPropertiesSet() throws Exception { - Assert.notNull(storeUri, "Property 'storeUri' is required"); - Assert.notNull(session, "Property 'JavaMailProperties' is required"); - Assert.notNull(converter, "An instantce of MailMessageConverter' is required"); - openSession(); - openFolder(); - } - - private void openFolder() throws MessagingException { - if (folder != null && folder.isOpen()) { - return; - } - folder = store.getFolder(storeUri); - if (folder == null || !folder.exists()) { - throw new IllegalStateException("No default folder to receive from"); - } - if (logger.isDebugEnabled()) { - logger.debug("Opening folder [" + MailTransportUtils.toPasswordProtectedString(storeUri) + "]"); - } - folder.open(monitoringStrategy.getFolderOpenMode()); - } - - private void openSession() throws MessagingException { - store = session.getStore(storeUri); - if (logger.isDebugEnabled()) { - logger.debug("Connecting to store [" + MailTransportUtils.toPasswordProtectedString(storeUri) + "]"); - } - store.connect(); - } - public void setConverter(MailMessageConverter converter) { this.converter = converter; } - public void destroy() throws Exception { - stop(); - } - - public boolean isRunning() { - return folder.isOpen(); - } - - public void start() { - try { - openSession(); - openFolder(); - } - catch (MessagingException messageE) { - throw new org.springframework.integration.message.MessagingException("Excpetion starting MailSource", - messageE); - } - - } - - public void stop() { - MailTransportUtils.closeFolder(folder); - MailTransportUtils.closeService(store); - } - public void setMailMessageStore(LocalMailMessageStore mailMessageStore) { this.mailMessageStore = mailMessageStore; } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java new file mode 100644 index 0000000000..48ccb2fb54 --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java @@ -0,0 +1,126 @@ +/* + * Copyright 2002-2007 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.adapter.mail; + +import javax.mail.Message; +import javax.mail.internet.MimeMessage; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.Lifecycle; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.SubscribableSource; + +/** + * Broadcasts all mail messages receoved to subscribed {@link MessageTarget} + * + * @author Jonas Partner + * + */ +public class SubscribableMailSource implements SubscribableSource, Lifecycle, + DisposableBean { + + private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(); + + private final TaskExecutor taskExecutor; + + private final MonitorRunnable monitorRunnable; + + private boolean monitorRunning = false; + + private final FolderConnection folderConnection; + + private MailMessageConverter converter = new DefaultMailMessageConverter(); + + public SubscribableMailSource(FolderConnection folderConnection, + TaskExecutor taskExecutor) { + this.folderConnection = folderConnection; + this.monitorRunnable = new MonitorRunnable(folderConnection); + this.taskExecutor = taskExecutor; + + } + + public void setApplySequence(boolean applySequence) { + this.dispatcher.setApplySequence(applySequence); + } + + public boolean subscribe(MessageTarget target) { + return this.dispatcher.addTarget(target); + } + + public boolean unsubscribe(MessageTarget target) { + return this.dispatcher.removeTarget(target); + } + + public void setConverter(MailMessageConverter converter) { + this.converter = converter; + } + + public void destroy() throws Exception { + stop(); + } + + public void start() { + startMonitor(); + } + + public void stop() { + stopMonitor(); + } + + public boolean isRunning() { + return monitorRunning; + } + + protected synchronized void startMonitor() { + if (!monitorRunning) { + taskExecutor.execute(monitorRunnable); + } + } + + protected synchronized void stopMonitor() { + if (monitorRunning) { + monitorRunnable.interrupt(); + } + } + + private class MonitorRunnable implements Runnable { + private volatile Thread thread; + + private final FolderConnection folderConnection; + + protected MonitorRunnable(FolderConnection folderConnection) { + this.folderConnection = folderConnection; + } + + public void interrupt() { + thread.interrupt(); + } + + public void run() { + thread = Thread.currentThread(); + while (!Thread.currentThread().isInterrupted()) { + Message[] messages = folderConnection.receive(); + for (Message message : messages) { + dispatcher.send(converter.create((MimeMessage) message)); + } + } + + } + } + +} diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/AsyncMonitoringStrategy.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/AsyncMonitoringStrategy.java new file mode 100644 index 0000000000..208351c70d --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/AsyncMonitoringStrategy.java @@ -0,0 +1,26 @@ +/* + * Copyright 2002-2007 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.adapter.mail.monitor; + +import javax.mail.Folder; +import javax.mail.MessagingException; + +public interface AsyncMonitoringStrategy { + + public abstract void waitForNewMessages(Folder folder) + throws MessagingException, InterruptedException; + +} \ No newline at end of file diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/DefaultLocalMailMessageStore.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/DefaultLocalMailMessageStore.java index 4bf606e146..16ddf00047 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/DefaultLocalMailMessageStore.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/DefaultLocalMailMessageStore.java @@ -35,6 +35,9 @@ public class DefaultLocalMailMessageStore implements LocalMailMessageStore { private ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); public void addLast(Message[] newMessages) { + if(newMessages == null){ + return; + } for (Message message : newMessages) { messages.add(message); } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/ImapIdleMonitoringStrategy.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/ImapIdleMonitoringStrategy.java index defbbbe52e..502ff9d875 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/ImapIdleMonitoringStrategy.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/monitor/ImapIdleMonitoringStrategy.java @@ -36,11 +36,14 @@ import com.sun.mail.imap.IMAPFolder; * * @author Arjen Poutsma */ -public class ImapIdleMonitoringStrategy extends AbstractMonitoringStrategy { +public class ImapIdleMonitoringStrategy extends AbstractMonitoringStrategy implements AsyncMonitoringStrategy { private MessageCountListener messageCountListener; - protected void waitForNewMessages(Folder folder) throws MessagingException, InterruptedException { + /* (non-Javadoc) + * @see org.springframework.integration.adapter.mail.monitor.AynchronouseMonitoringStrategy#waitForNewMessages(javax.mail.Folder) + */ + public void waitForNewMessages(Folder folder) throws MessagingException, InterruptedException { Assert.isInstanceOf(IMAPFolder.class, folder); IMAPFolder imapFolder = (IMAPFolder) folder; //retrieve unseen messages before we enter the blocking idle call diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/PollingMessageSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/PollingMessageSourceTests.java new file mode 100644 index 0000000000..a71dde172b --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/PollingMessageSourceTests.java @@ -0,0 +1,89 @@ +/* + * Copyright 2002-2007 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.adapter.mail; + +import static org.junit.Assert.*; + +import java.util.concurrent.ConcurrentLinkedQueue; + +import javax.mail.internet.MimeMessage; + +import org.easymock.classextension.EasyMock; +import org.junit.Test; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; + +public class PollingMessageSourceTests { + + @Test + public void testPolling(){ + StubFolderConnection folderConnection = new StubFolderConnection(); + + MimeMessage messageOne = EasyMock.createMock(MimeMessage.class); + MimeMessage messageTwo = EasyMock.createMock(MimeMessage.class); + MimeMessage messageThree = EasyMock.createMock(MimeMessage.class); + MimeMessage messageFour = EasyMock.createMock(MimeMessage.class); + + folderConnection.messages.add(new javax.mail.Message[]{messageOne}); + folderConnection.messages.add(new javax.mail.Message[]{messageTwo,messageThree}); + folderConnection.messages.add(new javax.mail.Message[]{messageFour}); + + PollingMailSource pollingMailSource = new PollingMailSource(folderConnection); + pollingMailSource.setConverter(new StubMessageConvertor()); + + + assertEquals("Wrong message for number 1", messageOne, pollingMailSource.receive().getPayload()); + assertEquals("Wrong message for number 2", messageTwo, pollingMailSource.receive().getPayload()); + assertEquals("Wrong message for number 3", messageThree, pollingMailSource.receive().getPayload()); + assertEquals("Wrong message for number 4", messageFour, pollingMailSource.receive().getPayload()); + assertNull("Expected null after exhausting all messages",pollingMailSource.receive()); + } + + + + private static class StubFolderConnection implements FolderConnection { + + ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); + + + + public javax.mail.Message[] receive() { + return messages.poll(); + } + + public boolean isRunning() { + return false; + } + + public void start() { + + } + + public void stop() { + + } + + } + + private static class StubMessageConvertor implements MailMessageConverter { + + @SuppressWarnings("unchecked") + public Message create(MimeMessage mailMessage) { + return new GenericMessage(mailMessage); + } + } + +} diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/SubscribableMailSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/SubscribableMailSourceTests.java new file mode 100644 index 0000000000..7b0015a6ce --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/SubscribableMailSourceTests.java @@ -0,0 +1,104 @@ +package org.springframework.integration.adapter.mail; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; + +import javax.mail.internet.MimeMessage; + +import org.easymock.classextension.EasyMock; +import org.junit.Before; +import org.junit.Test; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; +import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; + +/** + * + * @author Jonas Partner + * + */ +public class SubscribableMailSourceTests { + + TaskExecutor executor; + + @Before + public void setUp() { + executor = new ConcurrentTaskExecutor(); + } + + @Test + public void testReceive() throws Exception { + javax.mail.Message message = EasyMock.createMock(MimeMessage.class); + StubFolderConnection folderConnection = new StubFolderConnection( + message); + StubTarget target = new StubTarget(); + + SubscribableMailSource mailSource = new SubscribableMailSource( + folderConnection, executor); + mailSource.subscribe(target); + mailSource.setConverter(new StubMessageConvertor()); + + mailSource.start(); + Thread.sleep(1000); + mailSource.stop(); + + assertEquals("Wrong message count", 1, target.messages.size()); + assertEquals("Wrong payload", message, target.messages.get(0) + .getPayload()); + } + + private static class StubFolderConnection implements FolderConnection { + + ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); + + public StubFolderConnection(javax.mail.Message message) { + messages.add(message); + } + + public javax.mail.Message[] receive() { + javax.mail.Message msg = messages.poll(); + if (msg == null) { + return new javax.mail.Message[] {}; + } + return new javax.mail.Message[] { msg }; + } + + public boolean isRunning() { + return false; + } + + public void start() { + + } + + public void stop() { + + } + + } + + private static class StubTarget implements MessageTarget { + + List> messages = new ArrayList>(); + + public boolean send(Message message) { + messages.add(message); + return true; + } + + } + + private static class StubMessageConvertor implements MailMessageConverter { + + @SuppressWarnings("unchecked") + public Message create(MimeMessage mailMessage) { + return new GenericMessage(mailMessage); + } + } + +}