refacotred PollingMailSource and added SubscribableMailSource for use with IMAP idle

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();
+	}
+
+}

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();
+
+}

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;
 	}

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));
+				}
+			}
+
+		}
+	}
+
+}

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;
+
+}

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<Message> messages = new ConcurrentLinkedQueue<Message>();
 	
 	public void addLast(Message[] newMessages) {
+		if(newMessages == null){
+			return;
+		}
 		for (Message message : newMessages) {
 			messages.add(message);
 		}

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

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<javax.mail.Message[]> messages = new ConcurrentLinkedQueue<javax.mail.Message[]>();
+
+		
+		
+		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<MimeMessage>(mailMessage);
+		}
+	}
+
+}

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<javax.mail.Message> messages = new ConcurrentLinkedQueue<javax.mail.Message>();
+
+		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<Message<?>> messages = new ArrayList<Message<?>>();
+
+		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<MimeMessage>(mailMessage);
+		}
+	}
+
+}