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.
This commit is contained in:
Mark Fisher
@@ -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<MimeMessage> {
|
||||
|
||||
/**
|
||||
* 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<String,Object> mapToMessageHeaders(MimeMessage mailMessage) {
|
||||
try {
|
||||
Map<String, Object> headers = new HashMap<String, Object>();
|
||||
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];
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String, Object> header = headerMapper.mapToMessageHeaders(mailMessage);
|
||||
GenericMessage<Object> message = new GenericMessage<Object>(mailMessage.getContent(), header);
|
||||
return message;
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new MessagingException("Conversion of MailMessage failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
*/
|
||||
|
||||
@@ -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);
|
||||
|
||||
}
|
||||
@@ -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[]) {
|
||||
|
||||
@@ -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<javax.mail.Message> {
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
private final FolderConnection folderConnection;
|
||||
|
||||
private MailMessageConverter converter = new DefaultMailMessageConverter();
|
||||
private final Queue<javax.mail.Message> mailQueue = new ConcurrentLinkedQueue<javax.mail.Message>();
|
||||
|
||||
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<javax.mail.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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -40,7 +40,6 @@
|
||||
</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>
|
||||
|
||||
@@ -52,9 +51,9 @@
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:attribute name="id" type="xsd:string" use="required"/>
|
||||
<xsd:attribute name="channel" type="xsd:string"/>
|
||||
<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:attribute name="task-executor" type="xsd:string" use="optional"/>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
|
||||
@@ -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<Message> messages = new ConcurrentLinkedQueue<Message>();
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
}
|
||||
@@ -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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> headerMap = new HashMap<String, Object>();
|
||||
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<String, Object> headerMap = new HashMap<String, Object>();
|
||||
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<String, Object> headerMap = new HashMap<String, Object>();
|
||||
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")));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<javax.mail.Message[]> messages = new ConcurrentLinkedQueue<javax.mail.Message[]>();
|
||||
private final ConcurrentLinkedQueue<javax.mail.Message[]> messages = new ConcurrentLinkedQueue<javax.mail.Message[]>();
|
||||
|
||||
|
||||
|
||||
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<MimeMessage>(mailMessage);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<javax.mail.Message> messages = new ConcurrentLinkedQueue<javax.mail.Message>();
|
||||
private final ConcurrentLinkedQueue<javax.mail.Message> messages = new ConcurrentLinkedQueue<javax.mail.Message>();
|
||||
|
||||
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<MimeMessage>(mailMessage);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user