Added SubscribabelMailSource and namespace support for pollable and subscribable mail source

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd

@@ -227,6 +227,36 @@
 			<xsd:attribute name="password" type="xsd:string"/>
 		</xsd:complexType>
 	</xsd:element>
+	
+	<xsd:element name="polling-mail-source">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines a polling mail source.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:attribute name="id" type="xsd:string" use="required"/>
+			<xsd:attribute name="store-uri" type="xsd:string" use="required"/>
+			<xsd:attribute name="mail-convertor" type="xsd:string" use="optional"/>
+		</xsd:complexType>
+	</xsd:element>
+	
+	<xsd:element name="imap-idle-mail-source">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines a polling mail source.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:attribute name="id" type="xsd:string" use="required"/>
+			<xsd:attribute name="store-uri" type="xsd:string" use="required"/>
+			<xsd:attribute name="task-executor" type="xsd:string" use="required"/>
+			<xsd:attribute name="mail-convertor" type="xsd:string" use="optional"/>
+		</xsd:complexType>
+	</xsd:element>
+	
+	
+	
 
 	<xsd:element name="console-source">
 		<xsd:complexType>

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/DefaultFolderConnection.java

@@ -36,8 +36,11 @@ import org.springframework.integration.adapter.mail.monitor.MonitoringStrategy;
 import org.springframework.util.Assert;
 
 /**
+ * A Connection to a mail folder capable of retrieving mail by utilising the
+ * given instance of {@link MonitoringStrategy}
+ * 
  * @author Jonas Partner
- *
+ * 
  */
 public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 		DisposableBean, FolderConnection {
@@ -46,7 +49,7 @@ public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 
 	private final URLName storeUri;
 
-	private final Session session;
+	private Session session;
 
 	private final MonitoringStrategy monitoringStrategy;
 
@@ -56,14 +59,16 @@ public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 
 	private Folder folder;
 
-	public DefaultFolderConnection(String storeUri, Properties javaMailProperties,
+	private Properties javaMailProperties = new Properties();
+
+	public DefaultFolderConnection(String storeUri,
 			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)) {
+				&& monitoringStrategy.getClass().isAssignableFrom(
+						AsyncMonitoringStrategy.class)) {
 			throw new ConfigurationException(
 					"Folder connection requires an AsyncMonitoringStragey if polling is disabled");
 		}
@@ -72,21 +77,21 @@ public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 	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)
+	/*
+	 * (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");
+			start();
 		}
-	
+
 		try {
 			if (!polling) {
 				((AsyncMonitoringStrategy) monitoringStrategy)
@@ -105,7 +110,7 @@ public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 	}
 
 	public synchronized boolean isRunning() {
-		return (folder!= null && folder.isOpen());
+		return (folder != null && folder.isOpen());
 	}
 
 	public synchronized void start() {
@@ -143,6 +148,7 @@ public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 	}
 
 	private void openSession() throws MessagingException {
+		session = Session.getInstance(javaMailProperties);
 		store = session.getStore(storeUri);
 		if (logger.isDebugEnabled()) {
 			logger.debug("Connecting to store ["
@@ -152,4 +158,12 @@ public class DefaultFolderConnection implements Lifecycle, InitializingBean,
 		store.connect();
 	}
 
+	public Properties getJavaMailProperties() {
+		return javaMailProperties;
+	}
+
+	public void setJavaMailProperties(Properties javaMailProperties) {
+		this.javaMailProperties = javaMailProperties;
+	}
+
 }

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/PollingMailSource.java

@@ -81,4 +81,16 @@ public class PollingMailSource implements PollableSource {
 		this.mailMessageStore = mailMessageStore;
 	}
 
+	public FolderConnection getFolderConnection() {
+		return folderConnection;
+	}
+
+	public MailMessageConverter getConverter() {
+		return converter;
+	}
+
+	public LocalMailMessageStore getMailMessageStore() {
+		return mailMessageStore;
+	}
+
 }

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java

@@ -18,15 +18,20 @@ package org.springframework.integration.adapter.mail;
 import javax.mail.Message;
 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.core.task.TaskExecutor;
+import org.springframework.integration.adapter.mail.monitor.AsyncMonitoringStrategy;
 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}
+ * Broadcasts all mail messages recovered to subscribed {@link MessageTarget}
+ * The given {@link FolderConnection} should be using an
+ * {@link AsyncMonitoringStrategy} to retrieve mail
  * 
  * @author Jonas Partner
  * 
@@ -42,13 +47,12 @@ public class SubscribableMailSource implements SubscribableSource, Lifecycle,
 
 	private boolean monitorRunning = false;
 
-	private final FolderConnection folderConnection;
+	private final Log logger = LogFactory.getLog(getClass());
 
 	private MailMessageConverter converter = new DefaultMailMessageConverter();
 
 	public SubscribableMailSource(FolderConnection folderConnection,
 			TaskExecutor taskExecutor) {
-		this.folderConnection = folderConnection;
 		this.monitorRunnable = new MonitorRunnable(folderConnection);
 		this.taskExecutor = taskExecutor;
 
@@ -75,11 +79,16 @@ public class SubscribableMailSource implements SubscribableSource, Lifecycle,
 	}
 
 	public void start() {
+		logger.info("Starting to monitor mailbox");
 		startMonitor();
+		logger.info("Started to monitor mailbox");
+
 	}
 
 	public void stop() {
+		logger.info("Stopping monitoring of mailbox");
 		stopMonitor();
+		logger.info("Stopped monitoring mailbox");
 	}
 
 	public boolean isRunning() {
@@ -107,7 +116,7 @@ public class SubscribableMailSource implements SubscribableSource, Lifecycle,
 			this.folderConnection = folderConnection;
 		}
 
-		public void interrupt() {
+		public synchronized void interrupt() {
 			thread.interrupt();
 		}
 

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/config/PollingMailSourceParser.java

@@ -0,0 +1,93 @@
+/*
+ * 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.config;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.adapter.mail.DefaultFolderConnection;
+import org.springframework.integration.adapter.mail.PollingMailSource;
+import org.springframework.integration.adapter.mail.monitor.MonitoringStrategy;
+import org.springframework.integration.adapter.mail.monitor.PollingMonitoringStrategy;
+import org.springframework.integration.adapter.mail.monitor.Pop3PollingMonitoringStrategy;
+import org.springframework.util.StringUtils;
+import org.w3c.dom.Element;
+
+/**
+ * 
+ * @author Jonas Partner
+ * 
+ */
+public class PollingMailSourceParser extends AbstractSingleBeanDefinitionParser {
+
+	protected Class<?> getBeanClass(Element element) {
+		return PollingMailSource.class;
+	}
+
+	protected boolean shouldGenerateId() {
+		return false;
+	}
+
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	protected void doParse(Element element, ParserContext parserContext,
+			BeanDefinitionBuilder builder) {
+		String mailConvertorRef = element.getAttribute("convertor");
+		String uri = element.getAttribute("store-uri");
+		String propertiesRef = element.getAttribute("javaMailProperties");
+		if (!StringUtils.hasLength(uri)) {
+			throw new ConfigurationException("A store-uri is required");
+		}
+		BeanDefinitionBuilder folderConnectionBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(DefaultFolderConnection.class);
+		String storeType = uri.substring(0, 4).toLowerCase();
+		MonitoringStrategy monitoringStrategy = null;
+
+		if (storeType.equals("pop3")) {
+			monitoringStrategy = new Pop3PollingMonitoringStrategy();
+		} else if (storeType.equals("imap")) {
+			monitoringStrategy = new PollingMonitoringStrategy();
+		} else {
+			throw new ConfigurationException(
+					"No monitoring strategy for store-uri " + uri);
+		}
+
+		folderConnectionBuilder.addConstructorArgValue(uri);
+		folderConnectionBuilder.addConstructorArgValue(monitoringStrategy);
+		// set polling true
+		folderConnectionBuilder.addConstructorArgValue(true);
+
+		if (StringUtils.hasText(propertiesRef)) {
+			folderConnectionBuilder.addPropertyReference("javaMailProperties",
+					propertiesRef);
+		}
+		String folderConnectionName = parserContext.getReaderContext()
+				.registerWithGeneratedName(
+						folderConnectionBuilder.getBeanDefinition());
+
+		builder.addDependsOn(folderConnectionName);
+		builder.addConstructorArgValue(folderConnectionBuilder
+				.getBeanDefinition());
+
+		if (StringUtils.hasText(mailConvertorRef)) {
+			builder.addPropertyReference("convertor", mailConvertorRef);
+		}
+
+	}
+}

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/config/SubscribableImapIdleMailSourceParser.java

@@ -0,0 +1,88 @@
+/*
+ * 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.config;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.adapter.mail.DefaultFolderConnection;
+import org.springframework.integration.adapter.mail.SubscribableMailSource;
+import org.springframework.integration.adapter.mail.monitor.ImapIdleMonitoringStrategy;
+import org.springframework.util.StringUtils;
+import org.w3c.dom.Element;
+
+/**
+ * 
+ * @author Jonas Partner
+ *
+ */
+public class SubscribableImapIdleMailSourceParser extends
+		AbstractSingleBeanDefinitionParser {
+	protected Class<?> getBeanClass(Element element) {
+		return SubscribableMailSource.class;
+	}
+
+	protected boolean shouldGenerateId() {
+		return false;
+	}
+
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	protected void doParse(Element element, ParserContext parserContext,
+			BeanDefinitionBuilder builder) {
+		String mailConvertorRef = element.getAttribute("convertor");
+		String uri = element.getAttribute("store-uri");
+		String taskExecutorRef = element.getAttribute("task-executor");
+		String propertiesRef = element.getAttribute("javaMailProperties");
+		if (!StringUtils.hasLength(uri)) {
+			throw new ConfigurationException(
+					"A value for the store-uri attribue is required");
+		}
+		if (!StringUtils.hasLength(taskExecutorRef)) {
+			throw new ConfigurationException(
+					"A value for the task-executor attribute is required");
+		}
+
+		BeanDefinitionBuilder folderConnectionBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(DefaultFolderConnection.class);
+		String storeType = uri.substring(0, 4).toLowerCase();
+
+		if (!storeType.equals("imap")) {
+			throw new ConfigurationException(
+					"store-uri must start with imap for the imap idle source");
+		}
+		folderConnectionBuilder.addConstructorArgValue(uri);
+		folderConnectionBuilder
+				.addConstructorArgValue(new ImapIdleMonitoringStrategy());
+		// set polling false
+		folderConnectionBuilder.addConstructorArgValue(false);
+		if (StringUtils.hasText(propertiesRef)) {
+			folderConnectionBuilder.addPropertyReference("javaMailProperties",
+					propertiesRef);
+		}
+
+		builder.addConstructorArgValue(folderConnectionBuilder
+				.getBeanDefinition());
+		builder.addConstructorArgReference(taskExecutorRef);
+
+		if (StringUtils.hasText(mailConvertorRef)) {
+			builder.addPropertyReference("convertor", mailConvertorRef);
+		}
+	}
+}

org.springframework.integration.adapter/src/main/resources/META-INF/spring-integration.parsers

@@ -9,5 +9,7 @@ jms-gateway=org.springframework.integration.adapter.jms.config.JmsGatewayParser
 jms-source=org.springframework.integration.adapter.jms.config.JmsSourceParser
 jms-target=org.springframework.integration.adapter.jms.config.JmsTargetParser
 mail-target=org.springframework.integration.adapter.mail.config.MailTargetParser
+polling-mail-source=org.springframework.integration.adapter.mail.config.PollingMailSourceParser
+imap-idle-mail-source=org.springframework.integration.adapter.mail.config.SubscribableImapIdleMailSourceParser
 rmi-gateway=org.springframework.integration.adapter.rmi.config.RmiGatewayParser
 rmi-handler=org.springframework.integration.adapter.rmi.config.RmiHandlerParser

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/SubscribableMailSourceTests.java

@@ -20,7 +20,7 @@ import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
 /**
  * 
  * @author Jonas Partner
- *
+ * 
  */
 public class SubscribableMailSourceTests {
 

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/config/PollingMailSourceParserTests.java

@@ -0,0 +1,31 @@
+/*
+ * 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.config;
+
+import org.junit.Test;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.adapter.mail.PollingMailSource;
+
+public class PollingMailSourceParserTests {
+	
+	@Test
+	public void testPop3(){
+		ApplicationContext context = new ClassPathXmlApplicationContext("pollingMailSourceParserTests.xml", PollingMailSourceParserTests.class);
+		PollingMailSource mailSource = (PollingMailSource)context.getBean("pollingPop3");
+	}
+
+}

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/mail/config/pollingMailSourceParserTests.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+	<si:polling-mail-source id="pollingPop3" store-uri="pop3://mailtest:mailtest@ubuntuservervm/INBOX" />
+
+	
+</beans>