From 78c49157bdcdc936068b73617f587fdc4c7ad246 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 26 Sep 2008 16:03:01 +0000 Subject: [PATCH] Refactored inbound Mail adapters. Removed MailHeaderMapper and MessageConverter strategies. The Message payload will be a javax.mail.Message instance. Transformers will be used to convert the payload (e.g. Mail-to-String) and map header values. --- .../mail/AbstractMailHeaderMapper.java | 181 ----------------- .../mail/DefaultFolderConnection.java | 192 +++++++++--------- .../mail/DefaultMailMessageConverter.java | 48 ----- .../mail/DefaultMailMessageHeaderMapper.java | 56 ----- .../integration/mail/FolderConnection.java | 14 +- ...ilSource.java => ListeningMailSource.java} | 93 ++++----- .../integration/mail/MailHeaders.java | 4 +- .../mail/MailMessageConverter.java | 34 ---- .../mail/MailSendingMessageConsumer.java | 12 +- .../integration/mail/PollingMailSource.java | 83 ++++---- .../SubscribableImapIdleMailSourceParser.java | 59 +++--- .../config/spring-integration-mail-1.0.xsd | 5 +- .../monitor/DefaultLocalMailMessageStore.java | 52 ----- .../mail/monitor/LocalMailMessageStore.java | 32 --- .../DefaultMailMessageHeaderMapperTests.java | 155 -------------- .../mail/PollingMessageSourceTests.java | 69 +++---- .../mail/SubscribableMailSourceTests.java | 24 +-- 17 files changed, 241 insertions(+), 872 deletions(-) delete mode 100644 org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/AbstractMailHeaderMapper.java delete mode 100644 org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageConverter.java delete mode 100644 org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapper.java rename org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/{SubscribableMailSource.java => ListeningMailSource.java} (66%) delete mode 100644 org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailMessageConverter.java delete mode 100644 org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/DefaultLocalMailMessageStore.java delete mode 100644 org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/LocalMailMessageStore.java delete mode 100644 org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapperTests.java diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/AbstractMailHeaderMapper.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/AbstractMailHeaderMapper.java deleted file mode 100644 index ef51e3e414..0000000000 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/AbstractMailHeaderMapper.java +++ /dev/null @@ -1,181 +0,0 @@ -/* - * Copyright 2002-2008 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.mail; - -import java.util.HashMap; -import java.util.Map; - -import javax.mail.Address; -import javax.mail.Message.RecipientType; -import javax.mail.internet.MimeMessage; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.integration.adapter.MessageHeaderMapper; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHeaders; -import org.springframework.integration.message.MessagingException; -import org.springframework.mail.javamail.MimeMailMessage; - -/** - * @author Jonas Partner - */ -public abstract class AbstractMailHeaderMapper implements MessageHeaderMapper { - - /** - * Retrieve the subject of an e-mail message from an integration message. - * - * @param message the integration {@link Message} - * @return the e-mail message subject - */ - protected abstract String getSubject(MessageHeaders message); - - /** - * Retrieve the recipients list from an integration message. - * - * @param message the integration {@link Message} - * @return recipients list (TO) - */ - protected abstract String[] getTo(MessageHeaders message); - - /** - * Retrieve the CC recipients list from an integration message. - * - * @param message the integration {@link Message} - * @return CC recipients list (e-mail addresses) - */ - protected abstract String[] getCc(MessageHeaders message); - - /** - * Retrieve the BCC recipients list from an integration message. - * - * @param message the integration {@link Message} - * @return BCC recipients list (e-mail addresses) - */ - protected abstract String[] getBcc(MessageHeaders message); - - /** - * Retrieve the From: e-mail address from an integration message. - * - * @param message the integration {@link Message} - * @return the From: e-mail address - */ - protected abstract String getFrom(MessageHeaders message); - - /** - * Retrieve the Reply To: e-mail address from an integration message. - * - * @param message the integration {@link Message} - * @return the ReplyTo: e-mail address - */ - protected abstract String getReplyTo(MessageHeaders message); - - private final Log logger = LogFactory.getLog(this.getClass()); - - public void mapFromMessageHeaders(MessageHeaders headers, MimeMessage mailMessage) { - MimeMailMessage message = new MimeMailMessage(mailMessage); - - final String subject = getSubject(headers); - final String[] to = getTo(headers); - final String[] cc = getCc(headers); - final String[] bcc = getBcc(headers); - final String from = getFrom(headers); - final String replyTo = getReplyTo(headers); - if (subject != null) { - message.setSubject(subject); - } - else if (logger.isWarnEnabled()) { - logger.warn("no 'SUBJECT' property available for mail message"); - } - if (to != null) { - message.setTo(to); - } - else if (logger.isWarnEnabled()) { - logger.warn("no 'TO' property available for mail message"); - } - if (cc != null) { - message.setCc(cc); - } - if (bcc != null) { - message.setBcc(bcc); - } - if (from != null) { - message.setFrom(from); - } - else if (logger.isWarnEnabled()) { - logger.warn("no 'FROM' property available for mail message"); - } - if (replyTo != null) { - message.setReplyTo(replyTo); - } - } - - public Map mapToMessageHeaders(MimeMessage mailMessage) { - try { - Map headers = new HashMap(); - headers.put(MailHeaders.FROM, convertToString(mailMessage.getFrom())); - headers.put(MailHeaders.BCC, convertToStringArray(mailMessage.getRecipients(RecipientType.BCC))); - headers.put(MailHeaders.CC, convertToStringArray(mailMessage.getRecipients(RecipientType.CC))); - headers.put(MailHeaders.TO, convertToStringArray(mailMessage.getRecipients(RecipientType.TO))); - headers.put(MailHeaders.REPLY_TO, convertToString(mailMessage.getReplyTo())); - headers.put(MailHeaders.SUBJECT, mailMessage.getSubject()); - return headers; - } - catch (Exception e) { - throw new MessagingException("Conversion of MailMessage headers failed", e); - } - } - - protected String retrieveAsString(MessageHeaders headers, String key) { - Object value = headers.get(key); - return (value instanceof String) ? (String) value : null; - } - - protected String[] retrieveAsStringArray(MessageHeaders headers, String key) { - Object value = headers.get(key); - if (value instanceof String[]) { - return (String[]) value; - } - if (value instanceof String) { - return new String[] { (String) value }; - } - return null; - } - - private String convertToString(Address[] addresses) { - if (addresses == null || addresses.length == 0) { - return null; - } - if (addresses.length != 1) { - throw new IllegalStateException("expected a single value but received an Array"); - } - return addresses[0].toString(); - } - - private String[] convertToStringArray(Address[] addresses) { - if (addresses != null) { - String[] addressStrings = new String[addresses.length]; - for (int i = 0; i < addresses.length; i++) { - addressStrings[i] = addresses[i].toString(); - } - return addressStrings; - } - return new String[0]; - } - -} diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultFolderConnection.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultFolderConnection.java index b220b6a7de..14313fc677 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultFolderConnection.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultFolderConnection.java @@ -29,7 +29,6 @@ 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.mail.monitor.AsyncMonitoringStrategy; @@ -43,128 +42,119 @@ import org.springframework.util.Assert; * * @author Jonas Partner */ -public class DefaultFolderConnection implements Lifecycle, InitializingBean, - DisposableBean, FolderConnection { +public class DefaultFolderConnection implements Lifecycle, DisposableBean, FolderConnection { private final Log logger = LogFactory.getLog(this.getClass()); private final URLName storeUri; - private Session session; - private final MonitoringStrategy monitoringStrategy; private final boolean polling; - private Store store; + private volatile Session session; - private Folder folder; + private volatile Store store; - private Properties javaMailProperties = new Properties(); + private volatile Folder folder; - public DefaultFolderConnection(String storeUri, - MonitoringStrategy monitoringStrategy, boolean polling) { + private volatile Properties javaMailProperties = new Properties(); + + private volatile boolean running; + + private final Object lifecycleMonitor = new Object(); + + + public DefaultFolderConnection(String storeUri, MonitoringStrategy monitoringStrategy, boolean polling) { this.storeUri = new URLName(storeUri); this.monitoringStrategy = monitoringStrategy; this.polling = polling; - if (!polling - && monitoringStrategy.getClass().isAssignableFrom( - AsyncMonitoringStrategy.class)) { + Assert.notNull(storeUri, "storeUri is required"); + Assert.notNull(monitoringStrategy, "monitoringStrategy is required"); + if (!polling && monitoringStrategy.getClass().isAssignableFrom(AsyncMonitoringStrategy.class)) { throw new ConfigurationException( - "Folder connection requires an AsyncMonitoringStragey if polling is disabled"); + "Folder connection requires an AsyncMonitoringStrategy if polling is disabled."); } } - public void afterPropertiesSet() throws Exception { - - Assert.notNull(storeUri, "Property 'storeUri' is required"); - Assert.notNull(monitoringStrategy, - "An instantce of MonitoringStrategy' is required"); - // - } - - /* - * (non-Javadoc) - * - * @see org.springframework.integration.adapter.mail.FolderConnectionI#receive() - */ - public synchronized Message[] receive() { - if (!isRunning()) { - start(); - } - - 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 { - session = Session.getInstance(javaMailProperties); - store = session.getStore(storeUri); - if (logger.isDebugEnabled()) { - logger.debug("Connecting to store [" - + MailTransportUtils.toPasswordProtectedString(storeUri) - + "]"); - } - store.connect(); - } - - public Properties getJavaMailProperties() { - return javaMailProperties; - } public void setJavaMailProperties(Properties javaMailProperties) { this.javaMailProperties = javaMailProperties; } + public synchronized Message[] receive() { + try { + this.openFolder(); + if (!this.polling) { + ((AsyncMonitoringStrategy) this.monitoringStrategy).waitForNewMessages(this.folder); + } + return this.monitoringStrategy.receive(this.folder); + } + catch (Exception e) { + throw new org.springframework.integration.message.MessagingException( + "failure occurred while receiving from folder", e); + } + } + + public void destroy() throws Exception { + this.stop(); + } + + /* + * Lifecycle implementation + */ + + public boolean isRunning() { + synchronized (this.lifecycleMonitor) { + return this.running; + } + } + + public synchronized void start() { + synchronized (this.lifecycleMonitor) { + try { + this.openSession(); + this.openFolder(); + this.running = true; + } + catch (MessagingException e) { + throw new org.springframework.integration.message.MessagingException( + "Failed to start FolderConnection", e); + } + } + } + + public synchronized void stop() { + synchronized (this.lifecycleMonitor) { + MailTransportUtils.closeFolder(this.folder); + MailTransportUtils.closeService(this.store); + this.folder = null; + this.store = null; + this.running = false; + } + } + + private void openFolder() throws MessagingException { + this.folder = this.store.getFolder(this.storeUri); + if (this.folder == null || !this.folder.exists()) { + throw new IllegalStateException("no default folder available"); + } + if (this.folder.isOpen()) { + return; + } + if (logger.isDebugEnabled()) { + logger.debug("Opening folder [" + MailTransportUtils.toPasswordProtectedString(this.storeUri) + "]"); + } + this.folder.open(this.monitoringStrategy.getFolderOpenMode()); + } + + private void openSession() throws MessagingException { + this.session = Session.getInstance(this.javaMailProperties); + this.store = this.session.getStore(this.storeUri); + if (logger.isDebugEnabled()) { + logger.debug("Connecting to store [" + MailTransportUtils.toPasswordProtectedString(this.storeUri) + "]"); + } + this.store.connect(); + } + } diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageConverter.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageConverter.java deleted file mode 100644 index 14075368b8..0000000000 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageConverter.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.mail; - -import java.util.Map; - -import javax.mail.internet.MimeMessage; - -import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessagingException; - -/** - * - * @author Jonas Partner - * - */ -public class DefaultMailMessageConverter implements MailMessageConverter { - - private DefaultMailMessageHeaderMapper headerMapper = new DefaultMailMessageHeaderMapper(); - - @SuppressWarnings("unchecked") - public Message create(MimeMessage mailMessage) { - try { - Map header = headerMapper.mapToMessageHeaders(mailMessage); - GenericMessage message = new GenericMessage(mailMessage.getContent(), header); - return message; - } - catch (Exception e) { - throw new MessagingException("Conversion of MailMessage failed", e); - } - } - -} diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapper.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapper.java deleted file mode 100644 index 9ed2caf4a6..0000000000 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapper.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Copyright 2002-2008 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.mail; - -import org.springframework.integration.message.MessageHeaders; - -/** - * @author Jonas Partner - */ -public class DefaultMailMessageHeaderMapper extends AbstractMailHeaderMapper { - - @Override - protected String getSubject(MessageHeaders headers) { - return this.retrieveAsString(headers, MailHeaders.SUBJECT); - } - - @Override - protected String[] getTo(MessageHeaders headers) { - return this.retrieveAsStringArray(headers, MailHeaders.TO); - } - - @Override - protected String[] getCc(MessageHeaders headers) { - return this.retrieveAsStringArray(headers, MailHeaders.CC); - } - - @Override - protected String[] getBcc(MessageHeaders headers) { - return this.retrieveAsStringArray(headers, MailHeaders.BCC); - } - - @Override - protected String getFrom(MessageHeaders headers) { - return this.retrieveAsString(headers, MailHeaders.FROM); - } - - @Override - protected String getReplyTo(MessageHeaders headers) { - return this.retrieveAsString(headers, MailHeaders.REPLY_TO); - } - -} diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/FolderConnection.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/FolderConnection.java index 7eadb37698..3e2861ecff 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/FolderConnection.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/FolderConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2008 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.mail; import javax.mail.Folder; @@ -21,12 +22,13 @@ import javax.mail.Message; import org.springframework.context.Lifecycle; /** - * * Encapsulates state for a restartable connection to a {@link Folder} and ensures thread safety + * Encapsulates state for a restartable connection to a {@link Folder} and + * ensures thread safety. + * * @author Jonas Partner - * */ -public interface FolderConnection extends Lifecycle{ +public interface FolderConnection extends Lifecycle { - Message[] receive(); + Message[] receive(); -} \ No newline at end of file +} diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/SubscribableMailSource.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java similarity index 66% rename from org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/SubscribableMailSource.java rename to org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java index 416e3fd302..29cc75dbfb 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/SubscribableMailSource.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java @@ -17,7 +17,6 @@ package org.springframework.integration.mail; import javax.mail.Message; -import javax.mail.internet.MimeMessage; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -25,81 +24,68 @@ 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.channel.MessageChannel; +import org.springframework.integration.endpoint.AbstractMessageProducingEndpoint; import org.springframework.integration.mail.monitor.AsyncMonitoringStrategy; -import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessageBuilder; import org.springframework.util.Assert; /** - * An event-driven mail source that sends Spring Integration Messages to its output channel. - * The given {@link FolderConnection} should be using an {@link AsyncMonitoringStrategy} to - * retrieve mail. + * An event-driven mail source that sends Spring Integration Messages to its + * output channel. The Message payload will be the {@link javax.mail.Message} + * instance that was received. The given {@link FolderConnection} should be + * using an {@link AsyncMonitoringStrategy} to retrieve mail. * * @author Jonas Partner + * @author Mark Fisher */ -public class SubscribableMailSource implements MessageSource, Lifecycle, DisposableBean { +public class ListeningMailSource extends AbstractMessageProducingEndpoint implements Lifecycle, DisposableBean { private final Log logger = LogFactory.getLog(this.getClass()); - private volatile MessageChannel outputChannel; - - private final TaskExecutor taskExecutor; - private final MonitorRunnable monitorRunnable; private volatile boolean monitorRunning = false; - private volatile MailMessageConverter converter = new DefaultMailMessageConverter(); + private volatile TaskExecutor taskExecutor; - public SubscribableMailSource(FolderConnection folderConnection, TaskExecutor taskExecutor) { + public ListeningMailSource(FolderConnection folderConnection) { Assert.notNull(folderConnection, "FolderConnection must not be null"); - Assert.notNull(taskExecutor, "TaskExecutor must not be null"); this.monitorRunnable = new MonitorRunnable(folderConnection); + } + + + public void setTaskExecutor(TaskExecutor taskExecutor) { this.taskExecutor = taskExecutor; } - - public void setOutputChannel(MessageChannel outputChannel) { - this.outputChannel = outputChannel; - } - - public void setConverter(MailMessageConverter converter) { - this.converter = converter; - } - - public void destroy() throws Exception { - this.stop(); - } - - public void start() { - if (logger.isInfoEnabled()) { - logger.info("Starting to monitor mailbox"); - } - this.startMonitor(); - if (logger.isInfoEnabled()) { - logger.info("Started to monitor mailbox"); - } - - } - - public void stop() { - if (logger.isInfoEnabled()) { - logger.info("Stopping monitoring of mailbox"); - } - this.stopMonitor(); - if (logger.isInfoEnabled()) { - logger.info("Stopped monitoring mailbox"); - } - } + // Lifecycle implementation public boolean isRunning() { return this.monitorRunning; } + public void start() { + this.startMonitor(); + if (logger.isInfoEnabled()) { + logger.info("started monitoring mailbox"); + } + } + + public void stop() { + this.stopMonitor(); + if (logger.isInfoEnabled()) { + logger.info("stopped monitoring mailbox"); + } + } + protected void startMonitor() { synchronized (this.monitorRunnable) { if (!this.monitorRunning) { + if (this.taskExecutor == null) { + this.taskExecutor = this.getTaskScheduler(); + } + Assert.state(this.taskExecutor != null, "TaskExecutor is required"); this.taskExecutor.execute(this.monitorRunnable); } this.monitorRunning = true; @@ -115,6 +101,10 @@ public class SubscribableMailSource implements MessageSource, Lifecycle, Disposa } } + public void destroy() throws Exception { + this.stop(); + } + private class MonitorRunnable implements Runnable { @@ -135,9 +125,12 @@ public class SubscribableMailSource implements MessageSource, Lifecycle, Disposa public void run() { this.thread = Thread.currentThread(); while (!Thread.currentThread().isInterrupted()) { - Message[] messages = this.folderConnection.receive(); - for (Message message : messages) { - outputChannel.send(converter.create((MimeMessage) message)); + if (!this.folderConnection.isRunning()) { + this.folderConnection.start(); + } + Message[] mailMessages = this.folderConnection.receive(); + for (Message mailMessage : mailMessages) { + ListeningMailSource.this.sendMessage(MessageBuilder.withPayload(mailMessage).build()); } } } diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailHeaders.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailHeaders.java index 426a21de13..eced417a6a 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailHeaders.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailHeaders.java @@ -17,8 +17,8 @@ package org.springframework.integration.mail; /** - * Pre-defined names and prefixes to be used for setting and/or retrieving Mail attributes - * from/to integration Message Headers. + * Pre-defined header names to be used for setting and/or retrieving Mail + * Message attributes from/to integration Message Headers. * * @author Mark Fisher */ diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailMessageConverter.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailMessageConverter.java deleted file mode 100644 index c9d7f3f019..0000000000 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailMessageConverter.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * 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.mail; - -import javax.mail.internet.MimeMessage; - -import org.springframework.integration.message.Message; - -/** - * Converts a {@link MimeMessage} to a {@link Message} - * @author Jonas Partner - * - */ -public interface MailMessageConverter { - - - @SuppressWarnings("unchecked") - public Message create(MimeMessage mailMessage); - -} diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailSendingMessageConsumer.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailSendingMessageConsumer.java index c89b809be6..fc4b990a2a 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailSendingMessageConsumer.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailSendingMessageConsumer.java @@ -90,7 +90,7 @@ public class MailSendingMessageConsumer implements MessageConsumer { mailMessage = new SimpleMailMessage(); mailMessage.setText(message.getPayload().toString()); } - this.configureHeaderValues(mailMessage, message.getHeaders()); + this.applyHeadersToMailMessage(mailMessage, message.getHeaders()); return mailMessage; } @@ -114,20 +114,20 @@ public class MailSendingMessageConsumer implements MessageConsumer { } } - private void configureHeaderValues(MailMessage mailMessage, MessageHeaders headers) { + private void applyHeadersToMailMessage(MailMessage mailMessage, MessageHeaders headers) { String subject = headers.get(MailHeaders.SUBJECT, String.class); if (subject != null) { mailMessage.setSubject(subject); } - String[] to = this.retrieveAsStringArray(headers, MailHeaders.TO); + String[] to = this.retrieveHeaderValueAsStringArray(headers, MailHeaders.TO); if (to != null) { mailMessage.setTo(to); } - String[] cc = this.retrieveAsStringArray(headers, MailHeaders.CC); + String[] cc = this.retrieveHeaderValueAsStringArray(headers, MailHeaders.CC); if (cc != null) { mailMessage.setCc(cc); } - String[] bcc = this.retrieveAsStringArray(headers, MailHeaders.BCC); + String[] bcc = this.retrieveHeaderValueAsStringArray(headers, MailHeaders.BCC); if (bcc != null) { mailMessage.setBcc(bcc); } @@ -141,7 +141,7 @@ public class MailSendingMessageConsumer implements MessageConsumer { } } - private String[] retrieveAsStringArray(MessageHeaders headers, String key) { + private String[] retrieveHeaderValueAsStringArray(MessageHeaders headers, String key) { Object value = headers.get(key); if (value != null) { if (value instanceof String[]) { diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/PollingMailSource.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/PollingMailSource.java index 6551fda986..4a67eefda3 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/PollingMailSource.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/PollingMailSource.java @@ -16,82 +16,67 @@ package org.springframework.integration.mail; -import javax.mail.internet.MimeMessage; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.mail.monitor.DefaultLocalMailMessageStore; -import org.springframework.integration.mail.monitor.LocalMailMessageStore; import org.springframework.integration.mail.monitor.MonitoringStrategy; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.PollableSource; +import org.springframework.util.Assert; /** - * {@link MessageSource} implementation which delegates to a + * {@link MessageSource} implementation that delegates to a * {@link MonitoringStrategy} to poll a mailbox. Each poll of the mailbox may - * return more than one message which will then be stored locally using the - * provided {@link LocalMailMessageStore} + * return more than one message which will then be stored in a queue. * * @author Jonas Partner + * @author Mark Fisher */ -@SuppressWarnings("unchecked") -public class PollingMailSource implements PollableSource { +public class PollingMailSource implements PollableSource { private final Log logger = LogFactory.getLog(this.getClass()); private final FolderConnection folderConnection; - private MailMessageConverter converter = new DefaultMailMessageConverter(); + private final Queue mailQueue = new ConcurrentLinkedQueue(); - private LocalMailMessageStore mailMessageStore = new DefaultLocalMailMessageStore(); - public PollingMailSource(FolderConnection folderConnetion) { - this.folderConnection = folderConnetion; + public PollingMailSource(FolderConnection folderConnection) { + Assert.notNull(folderConnection, "folderConnection must not be null"); + this.folderConnection = folderConnection; } + @SuppressWarnings("unchecked") - public Message receive() { - Message received = null; - javax.mail.Message mailMessage = mailMessageStore.getNext(); - if (mailMessage == null) { - try { - 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); + public Message receive() { + try { + javax.mail.Message mailMessage = this.mailQueue.poll(); + if (mailMessage == null) { + javax.mail.Message[] messages = this.folderConnection.receive(); + if (messages != null) { + for (javax.mail.Message message : messages) { + this.mailQueue.add(message); + } + } + mailMessage = this.mailQueue.poll(); + } + if (mailMessage != null) { + if (logger.isDebugEnabled()) { + logger.debug("Received mail message [" + mailMessage + "]"); + } + return MessageBuilder.withPayload(mailMessage).build(); } } - if (mailMessage != null) { - received = converter.create((MimeMessage) mailMessage); - if (logger.isDebugEnabled()) { - logger.debug("Received message " + received); - } + catch (Exception e) { + throw new MessagingException("failure occurred while polling for mail", e); } - return received; - } - - public void setConverter(MailMessageConverter converter) { - this.converter = converter; - } - - public void setMailMessageStore(LocalMailMessageStore mailMessageStore) { - this.mailMessageStore = mailMessageStore; - } - - public FolderConnection getFolderConnection() { - return folderConnection; - } - - public MailMessageConverter getConverter() { - return converter; - } - - public LocalMailMessageStore getMailMessageStore() { - return mailMessageStore; + return null; } } diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/SubscribableImapIdleMailSourceParser.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/SubscribableImapIdleMailSourceParser.java index d9537693e4..d54c15e198 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/SubscribableImapIdleMailSourceParser.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/SubscribableImapIdleMailSourceParser.java @@ -21,19 +21,22 @@ import org.w3c.dom.Element; 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.mail.DefaultFolderConnection; -import org.springframework.integration.mail.SubscribableMailSource; +import org.springframework.integration.mail.ListeningMailSource; import org.springframework.integration.mail.monitor.ImapIdleMonitoringStrategy; +import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** + * Parser for the <imap-idle-mail-source> element in the 'mail' namespace. + * * @author Jonas Partner + * @author Mark Fisher */ -public class SubscribableImapIdleMailSourceParser extends - AbstractSingleBeanDefinitionParser { +public class SubscribableImapIdleMailSourceParser extends AbstractSingleBeanDefinitionParser { + protected Class getBeanClass(Element element) { - return SubscribableMailSource.class; + return ListeningMailSource.class; } protected boolean shouldGenerateId() { @@ -44,45 +47,29 @@ public class SubscribableImapIdleMailSourceParser extends return true; } - protected void doParse(Element element, ParserContext parserContext, - BeanDefinitionBuilder builder) { - String mailConvertorRef = element.getAttribute("convertor"); + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + String channel = element.getAttribute("channel"); 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"); - } + Assert.hasText(channel, "the 'channel' attribute is required"); + Assert.hasText(uri, "the 'store-uri' attribute is required"); + BeanDefinitionBuilder folderConnectionBuilder = + BeanDefinitionBuilder.genericBeanDefinition(DefaultFolderConnection.class); + Assert.isTrue(uri.toLowerCase().startsWith("imap"), + "store-uri must start with 'imap' for the imap idle source"); folderConnectionBuilder.addConstructorArgValue(uri); - folderConnectionBuilder - .addConstructorArgValue(new ImapIdleMonitoringStrategy()); + folderConnectionBuilder.addConstructorArgValue(new ImapIdleMonitoringStrategy()); // set polling false folderConnectionBuilder.addConstructorArgValue(false); if (StringUtils.hasText(propertiesRef)) { - folderConnectionBuilder.addPropertyReference("javaMailProperties", - propertiesRef); + folderConnectionBuilder.addPropertyReference("javaMailProperties", propertiesRef); } - - builder.addConstructorArgValue(folderConnectionBuilder - .getBeanDefinition()); - builder.addConstructorArgReference(taskExecutorRef); - - if (StringUtils.hasText(mailConvertorRef)) { - builder.addPropertyReference("convertor", mailConvertorRef); + builder.addConstructorArgValue(folderConnectionBuilder.getBeanDefinition()); + if (StringUtils.hasLength(taskExecutorRef)) { + builder.addPropertyReference("taskExecutor", taskExecutorRef); } + builder.addPropertyReference("outputChannel", channel); } + } diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/spring-integration-mail-1.0.xsd b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/spring-integration-mail-1.0.xsd index 8fd57381f3..ae2c9c643c 100644 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/spring-integration-mail-1.0.xsd +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/spring-integration-mail-1.0.xsd @@ -40,7 +40,6 @@ - @@ -52,9 +51,9 @@ + - - + diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/DefaultLocalMailMessageStore.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/DefaultLocalMailMessageStore.java deleted file mode 100644 index f5f260c45c..0000000000 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/DefaultLocalMailMessageStore.java +++ /dev/null @@ -1,52 +0,0 @@ -/* - * 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.mail.monitor; - -import java.util.concurrent.ConcurrentLinkedQueue; - -import javax.mail.Message; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -/** - * - * @author Jonas Partner - * - */ -public class DefaultLocalMailMessageStore implements LocalMailMessageStore { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); - - public void addLast(Message[] newMessages) { - if(newMessages == null){ - return; - } - for (Message message : newMessages) { - messages.add(message); - } - logger.info("LocalMailMessageStore size is now" + messages.size()); - } - - public Message getNext() { - logger.info("Message store size " + messages.size()); - return messages.poll(); - } - -} diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/LocalMailMessageStore.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/LocalMailMessageStore.java deleted file mode 100644 index c43663ee5b..0000000000 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/LocalMailMessageStore.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * 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.mail.monitor; - -import javax.mail.Message; - -/** - * Acts as a buffer for downloaded MailMessages - * @author Jonas Partner - * - */ -public interface LocalMailMessageStore { - - public Message getNext(); - - public void addLast(Message[] newMessages); - -} diff --git a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapperTests.java b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapperTests.java deleted file mode 100644 index 8c718cf307..0000000000 --- a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/DefaultMailMessageHeaderMapperTests.java +++ /dev/null @@ -1,155 +0,0 @@ -/* - * Copyright 2002-2008 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.mail; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; - -import java.util.HashMap; -import java.util.Map; - -import javax.mail.Address; -import javax.mail.MessagingException; -import javax.mail.Message.RecipientType; -import javax.mail.internet.InternetAddress; -import javax.mail.internet.MimeMessage; - -import org.easymock.classextension.EasyMock; -import org.junit.Test; - -import org.springframework.integration.mail.DefaultMailMessageHeaderMapper; -import org.springframework.integration.mail.MailHeaders; -import org.springframework.integration.message.MessageHeaders; -import org.springframework.util.ObjectUtils; - -/** - * @author Mark Fisher - */ -public class DefaultMailMessageHeaderMapperTests { - - @Test - public void mapExactlyOneFromAttributeFromMimeMessage() throws MessagingException { - DefaultMailMessageHeaderMapper mapper = new DefaultMailMessageHeaderMapper(); - MimeMessage mailMessageMock = EasyMock.createMock(MimeMessage.class); - Address[] fromAddresses = new Address[] { new InternetAddress("from@example.org") }; - EasyMock.expect(mailMessageMock.getFrom()).andReturn(fromAddresses); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.BCC)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.CC)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.TO)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getReplyTo()).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getSubject()).andReturn("mail test"); - EasyMock.replay(mailMessageMock); - Map headers = mapper.mapToMessageHeaders(mailMessageMock); - Object fromHeader = headers.get(MailHeaders.FROM); - assertNotNull(fromHeader); - assertTrue(fromHeader instanceof String); - assertEquals("from@example.org", fromHeader); - } - - @Test - public void mapExactlyOneReplyToAttributeFromMimeMessage() throws MessagingException { - DefaultMailMessageHeaderMapper mapper = new DefaultMailMessageHeaderMapper(); - MimeMessage mailMessageMock = EasyMock.createMock(MimeMessage.class); - Address[] replyToAddresses = new Address[] { new InternetAddress("replyTo@example.org") }; - EasyMock.expect(mailMessageMock.getFrom()).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.BCC)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.CC)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.TO)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getReplyTo()).andReturn(replyToAddresses); - EasyMock.expect(mailMessageMock.getSubject()).andReturn("mail test"); - EasyMock.replay(mailMessageMock); - Map headers = mapper.mapToMessageHeaders(mailMessageMock); - Object replyToHeader = headers.get(MailHeaders.REPLY_TO); - assertNotNull(replyToHeader); - assertTrue(replyToHeader instanceof String); - assertEquals("replyTo@example.org", replyToHeader); - } - - @Test - public void mapMultipleToAttributesFromMimeMessage() throws MessagingException { - DefaultMailMessageHeaderMapper mapper = new DefaultMailMessageHeaderMapper(); - MimeMessage mailMessageMock = EasyMock.createMock(MimeMessage.class); - Address[] toAddresses = new Address[] { - new InternetAddress("a@example.org"), new InternetAddress("b@example.org"), new InternetAddress("c@example.org") - }; - EasyMock.expect(mailMessageMock.getFrom()).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.BCC)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.CC)).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getRecipients(RecipientType.TO)).andReturn(toAddresses); - EasyMock.expect(mailMessageMock.getReplyTo()).andReturn(new Address[0]); - EasyMock.expect(mailMessageMock.getSubject()).andReturn("mail test"); - EasyMock.replay(mailMessageMock); - Map headers = mapper.mapToMessageHeaders(mailMessageMock); - Object toHeader = headers.get(MailHeaders.TO); - assertNotNull(toHeader); - assertTrue(toHeader instanceof String[]); - String[] addresses = (String[]) toHeader; - assertTrue(ObjectUtils.containsElement(addresses, "a@example.org")); - assertTrue(ObjectUtils.containsElement(addresses, "b@example.org")); - assertTrue(ObjectUtils.containsElement(addresses, "c@example.org")); - } - - @Test - public void mapReplyToValueFromHeadersToMimeMessage() throws MessagingException { - DefaultMailMessageHeaderMapper mapper = new DefaultMailMessageHeaderMapper(); - Map headerMap = new HashMap(); - headerMap.put(MailHeaders.REPLY_TO, "replyTo@example.org"); - MessageHeaders headers = new MessageHeaders(headerMap); - MimeMessage mailMessageMock = EasyMock.createNiceMock(MimeMessage.class); - EasyMock.replay(mailMessageMock); - MimeMessage mimeMessage = new MimeMessage(mailMessageMock); - mapper.mapFromMessageHeaders(headers, mimeMessage); - Address[] replyToAddresses = mimeMessage.getReplyTo(); - assertEquals(1, replyToAddresses.length); - assertEquals("replyTo@example.org", replyToAddresses[0].toString()); - } - - @Test - public void mapFromValueFromHeadersToMimeMessage() throws MessagingException { - DefaultMailMessageHeaderMapper mapper = new DefaultMailMessageHeaderMapper(); - Map headerMap = new HashMap(); - headerMap.put(MailHeaders.FROM, "from@example.org"); - MessageHeaders headers = new MessageHeaders(headerMap); - MimeMessage mailMessageMock = EasyMock.createNiceMock(MimeMessage.class); - EasyMock.replay(mailMessageMock); - MimeMessage mimeMessage = new MimeMessage(mailMessageMock); - mapper.mapFromMessageHeaders(headers, mimeMessage); - Address[] fromAddresses = mimeMessage.getFrom(); - assertEquals(1, fromAddresses.length); - assertEquals("from@example.org", fromAddresses[0].toString()); - } - - @Test - public void mapMultileToValuesFromHeadersToMimeMessage() throws MessagingException { - DefaultMailMessageHeaderMapper mapper = new DefaultMailMessageHeaderMapper(); - Map headerMap = new HashMap(); - String[] addressStrings = new String[] { "a@example.org", "b@example.org", "c@example.org" }; - headerMap.put(MailHeaders.TO, addressStrings); - MessageHeaders headers = new MessageHeaders(headerMap); - MimeMessage mailMessageMock = EasyMock.createNiceMock(MimeMessage.class); - EasyMock.replay(mailMessageMock); - MimeMessage mimeMessage = new MimeMessage(mailMessageMock); - mapper.mapFromMessageHeaders(headers, mimeMessage); - Address[] toAddresses = mimeMessage.getRecipients(RecipientType.TO); - assertEquals(3, toAddresses.length); - assertTrue(ObjectUtils.containsElement(toAddresses, new InternetAddress("a@example.org"))); - assertTrue(ObjectUtils.containsElement(toAddresses, new InternetAddress("b@example.org"))); - assertTrue(ObjectUtils.containsElement(toAddresses, new InternetAddress("c@example.org"))); - } - -} diff --git a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/PollingMessageSourceTests.java b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/PollingMessageSourceTests.java index c93d156112..c57fc1311c 100644 --- a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/PollingMessageSourceTests.java +++ b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/PollingMessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2008 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. @@ -13,9 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.mail; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import java.util.concurrent.ConcurrentLinkedQueue; @@ -24,46 +26,36 @@ import javax.mail.internet.MimeMessage; import org.easymock.classextension.EasyMock; import org.junit.Test; -import org.springframework.integration.mail.FolderConnection; -import org.springframework.integration.mail.MailMessageConverter; -import org.springframework.integration.mail.PollingMailSource; -import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.Message; - +/** + * @author Jonas Partner + */ public class PollingMessageSourceTests { @Test - public void testPolling(){ + 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}); - + MimeMessage message1 = EasyMock.createMock(MimeMessage.class); + MimeMessage message2 = EasyMock.createMock(MimeMessage.class); + MimeMessage message3 = EasyMock.createMock(MimeMessage.class); + MimeMessage message4 = EasyMock.createMock(MimeMessage.class); + + folderConnection.messages.add(new javax.mail.Message[] { message1 }); + folderConnection.messages.add(new javax.mail.Message[] { message2, message3 }); + folderConnection.messages.add(new javax.mail.Message[] { message4 }); + 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()); + assertEquals("Wrong message for number 1", message1, pollingMailSource.receive().getPayload()); + assertEquals("Wrong message for number 2", message2, pollingMailSource.receive().getPayload()); + assertEquals("Wrong message for number 3", message3, pollingMailSource.receive().getPayload()); + assertEquals("Wrong message for number 4", message4, pollingMailSource.receive().getPayload()); + assertNull("Expected null after exhausting all messages", pollingMailSource.receive()); } - - - + + private static class StubFolderConnection implements FolderConnection { - ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); + private final ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); - - public javax.mail.Message[] receive() { return messages.poll(); } @@ -73,20 +65,9 @@ public class PollingMessageSourceTests { } 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.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java index b0ae6d6474..80e3965f1e 100644 --- a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java +++ b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java @@ -26,12 +26,9 @@ 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.channel.QueueChannel; -import org.springframework.integration.mail.FolderConnection; -import org.springframework.integration.mail.MailMessageConverter; -import org.springframework.integration.mail.SubscribableMailSource; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; @@ -40,21 +37,23 @@ import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; */ public class SubscribableMailSourceTests { - TaskExecutor executor; + private 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); QueueChannel channel = new QueueChannel(); - SubscribableMailSource mailSource = new SubscribableMailSource(folderConnection, executor); + ListeningMailSource mailSource = new ListeningMailSource(folderConnection); + mailSource.setTaskExecutor(executor); mailSource.setOutputChannel(channel); - mailSource.setConverter(new StubMessageConvertor()); mailSource.start(); Message result = channel.receive(1000); mailSource.stop(); @@ -65,7 +64,7 @@ public class SubscribableMailSourceTests { private static class StubFolderConnection implements FolderConnection { - ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); + private final ConcurrentLinkedQueue messages = new ConcurrentLinkedQueue(); public StubFolderConnection(javax.mail.Message message) { messages.add(message); @@ -90,13 +89,4 @@ public class SubscribableMailSourceTests { } } - - private static class StubMessageConvertor implements MailMessageConverter { - - @SuppressWarnings("unchecked") - public Message create(MimeMessage mailMessage) { - return new GenericMessage(mailMessage); - } - } - }