Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactored inbound mail channel adapters. Refactored the classes from the 'monitor' package to include a MailReceiver strategy interface with POP3 and IMAP implementations. Added a polling inbound channel adapter that polls a MailReceivingMessageSource which in turn delegates to one of the MailReceiver implementations. Also, added the ImapIdleChannelAdapter for asynchronous callbacks rather than polling (INT-444).

Browse Source
This commit is contained in:
Mark Fisher
2008-10-30 19:52:47 +00:00
parent 1631fe3921
commit 3817fa2598
26 changed files with 756 additions and 1019 deletions
Download Patch File Download Diff File Expand all files Collapse all files

217
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java Executable file
View File

@@ -0,0 +1,217 @@
/*
* 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.Properties;
import javax.mail.FetchProfile;
import javax.mail.Flags;
import javax.mail.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.Session;
import javax.mail.Store;
import javax.mail.URLName;
import javax.mail.internet.MimeMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.mail.monitor.MailTransportUtils;
import org.springframework.util.Assert;
/**
* Base class for {@link MailReceiver} implementations.
*
* @author Arjen Poutsma
* @author Jonas Partner
* @author Mark Fisher
*/
public abstract class AbstractMailReceiver implements MailReceiver, DisposableBean {
protected final Log logger = LogFactory.getLog(this.getClass());
private final URLName url;
private volatile int maxFetchSize = -1;
private volatile Session session;
private volatile Store store;
private volatile Folder folder;
private volatile Properties javaMailProperties = new Properties();
protected volatile boolean initialized;
private final Object initializationMonitor = new Object();
public AbstractMailReceiver(URLName urlName) {
Assert.notNull(urlName, "urlName must not be null");
this.url = urlName;
}
public AbstractMailReceiver(String url) {
Assert.notNull(url, "url must not be null");
this.url = new URLName(url);
}
public void setJavaMailProperties(Properties javaMailProperties) {
this.javaMailProperties = javaMailProperties;
}
public void setMaxFetchSize(int maxFetchSize) {
this.maxFetchSize = maxFetchSize;
}
protected Folder getFolder() {
return this.folder;
}
/**
* Subclasses must implement this method to return new mail messages.
*/
protected abstract Message[] searchForNewMessages() throws MessagingException;
/**
* Subclasses must implement this method to indicate whether the mail
* messages should be deleted after being received.
*/
protected abstract boolean shouldDeleteMessages();
private void openSession() throws MessagingException {
if (this.session == null) {
this.session = Session.getInstance(this.javaMailProperties);
}
if (this.store == null) {
this.store = this.session.getStore(this.url);
}
if (!this.store.isConnected()) {
if (logger.isDebugEnabled()) {
logger.debug("connecting to store [" + MailTransportUtils.toPasswordProtectedString(this.url) + "]");
}
this.store.connect();
}
}
protected void openFolder() throws MessagingException {
this.openSession();
if (this.folder == null) {
this.folder = this.store.getFolder(this.url);
}
if (this.folder == null || !this.folder.exists()) {
throw new IllegalStateException("no such folder [" + this.url.getFile() + "]");
}
if (this.folder.isOpen()) {
return;
}
if (logger.isDebugEnabled()) {
logger.debug("opening folder [" + MailTransportUtils.toPasswordProtectedString(this.url) + "]");
}
if (this.shouldDeleteMessages()) {
this.folder.open(Folder.READ_WRITE);
}
else {
this.folder.open(Folder.READ_ONLY);
}
}
public synchronized Message[] receive() {
try {
this.openFolder();
if (logger.isInfoEnabled()) {
logger.info("attempting to receive mail from folder [" + this.getFolder().getFullName() + "]");
}
Message[] messages = this.searchForNewMessages();
if (this.maxFetchSize > 0 && messages.length > this.maxFetchSize) {
Message[] reducedMessages = new Message[this.maxFetchSize];
System.arraycopy(messages, 0, reducedMessages, 0, this.maxFetchSize);
messages = reducedMessages;
}
if (logger.isDebugEnabled()) {
logger.debug("found " + messages.length + " new messages");
}
if (messages.length > 0) {
this.fetchMessages(messages);
}
if (this.shouldDeleteMessages()) {
this.deleteMessages(messages);
}
Message[] copiedMessages = new Message[messages.length];
for (int i = 0; i < messages.length; i++) {
copiedMessages[i] = new MimeMessage((MimeMessage) messages[i]);
}
return copiedMessages;
}
catch (Exception e) {
throw new org.springframework.integration.core.MessagingException(
"failure occurred while receiving from folder", e);
}
finally {
MailTransportUtils.closeFolder(this.folder);
}
}
/**
* Fetches the specified messages from this receiver's folder. Default
* implementation {@link Folder#fetch(Message[], FetchProfile) fetches}
* every {@link javax.mail.FetchProfile.Item}.
*
* @param messages the messages to fetch
* @throws MessagingException in case of JavMail errors
*/
protected void fetchMessages(Message[] messages) throws MessagingException {
FetchProfile contentsProfile = new FetchProfile();
contentsProfile.add(FetchProfile.Item.ENVELOPE);
contentsProfile.add(FetchProfile.Item.CONTENT_INFO);
contentsProfile.add(FetchProfile.Item.FLAGS);
this.folder.fetch(messages, contentsProfile);
}
/**
* Deletes the given messages from this receiver's folder. Only invoked
* when {@link #setDeleteMessages(boolean)} is <code>true</code>.
*
* @param messages the messages to delete
* @throws MessagingException in case of JavaMail errors
*/
protected void deleteMessages(Message[] messages) throws MessagingException {
for (int i = 0; i < messages.length; i++) {
messages[i].setFlag(Flags.Flag.DELETED, true);
}
}
public void destroy() throws Exception {
synchronized (this.initializationMonitor) {
MailTransportUtils.closeFolder(this.folder);
MailTransportUtils.closeService(this.store);
this.folder = null;
this.store = null;
this.initialized = false;
}
}
@Override
public String toString() {
return this.url.toString();
}
}

164
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/DefaultFolderConnection.java
View File

@@ -1,164 +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.Properties;
import javax.mail.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.Session;
import javax.mail.Store;
import javax.mail.URLName;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.mail.monitor.AsyncMonitoringStrategy;
import org.springframework.integration.mail.monitor.MailTransportUtils;
import org.springframework.integration.mail.monitor.MonitoringStrategy;
import org.springframework.util.Assert;
/**
* A Connection to a mail folder capable of retrieving mail by utilizing the
* given instance of {@link MonitoringStrategy}.
*
* @author Jonas Partner
*/
public class DefaultFolderConnection implements Lifecycle, DisposableBean, FolderConnection {
private final Log logger = LogFactory.getLog(this.getClass());
private final URLName storeUri;
private final MonitoringStrategy monitoringStrategy;
private final boolean polling;
private volatile Session session;
private volatile Store store;
private volatile Folder folder;
private volatile Properties javaMailProperties = new Properties();
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
public DefaultFolderConnection(String storeUri, MonitoringStrategy monitoringStrategy, boolean polling) {
Assert.notNull(storeUri, "storeUri must not be null");
Assert.notNull(monitoringStrategy, "monitoringStrategy must not ne null");
this.storeUri = new URLName(storeUri);
this.monitoringStrategy = monitoringStrategy;
this.polling = polling;
Assert.isTrue(polling || AsyncMonitoringStrategy.class.isAssignableFrom(monitoringStrategy.getClass()),
"Folder connection requires an AsyncMonitoringStrategy if polling is disabled.");
}
public void setJavaMailProperties(Properties javaMailProperties) {
this.javaMailProperties = javaMailProperties;
}
public synchronized Message[] receive() {
try {
if (!this.isRunning()) {
this.start();
}
if (!this.polling) {
((AsyncMonitoringStrategy) this.monitoringStrategy).waitForNewMessages(this.folder);
}
return this.monitoringStrategy.receive(this.folder);
}
catch (Exception e) {
throw new org.springframework.integration.core.MessagingException(
"failure occurred while receiving from folder", e);
}
}
@Override
public String toString() {
return this.storeUri.toString();
}
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.core.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();
}
}

160
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java Executable file
View File

@@ -0,0 +1,160 @@
/*
* 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.Properties;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.endpoint.AbstractMessageProducingEndpoint;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.Assert;
/**
* An event-driven Channel Adapter that receives mail messages from a mail
* server that supports the IMAP "idle" command (see RFC 2177). Received mail
* messages will be converted and sent as Spring Integration Messages to the
* output channel. The Message payload will be the {@link javax.mail.Message}
* instance that was received.
*
* @author Arjen Poutsma
* @author Mark Fisher
*/
public class ImapIdleChannelAdapter extends AbstractMessageProducingEndpoint implements Lifecycle {
private final IdleTask idleTask = new IdleTask();
private volatile TaskExecutor taskExecutor;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
private final ImapMailReceiver mailReceiver;
public ImapIdleChannelAdapter(ImapMailReceiver mailReceiver) {
Assert.notNull(mailReceiver, "mailReceiver must not be null");
this.mailReceiver = mailReceiver;
}
public ImapIdleChannelAdapter(String url) {
Assert.isTrue(url.startsWith("imap"), "url must start with 'imap'");
this.mailReceiver = new ImapMailReceiver(url);
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void setJavaMailProperties(Properties javaMailProperties) {
this.mailReceiver.setJavaMailProperties(javaMailProperties);
}
public void setShouldDeleteMessages(boolean shouldDeleteMessages) {
this.mailReceiver.setShouldDeleteMessages(shouldDeleteMessages);
}
protected void handleMailMessagingException(MessagingException e) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred in idle task", e);
}
}
/*
* Lifecycle implementation
*/
public boolean isRunning() {
return this.running;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
if (this.taskExecutor == null) {
if (logger.isInfoEnabled()) {
logger.info("No TaskExecutor has been provided, will use a ["
+ SimpleAsyncTaskExecutor.class + "] as the default.");
}
this.taskExecutor = new SimpleAsyncTaskExecutor();
}
this.taskExecutor.execute(this.idleTask);
}
this.running = true;
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.running) {
this.idleTask.interrupt();
}
this.running = false;
}
}
private class IdleTask implements Runnable {
private volatile Thread thread;
public synchronized void interrupt() {
if (this.thread != null) {
this.thread.interrupt();
}
else if (logger.isInfoEnabled()) {
logger.info("monitor is not running, cannot interrupt");
}
}
public void run() {
this.thread = Thread.currentThread();
while (!Thread.currentThread().isInterrupted()) {
try {
if (logger.isDebugEnabled()) {
logger.debug("waiting for mail");
}
mailReceiver.waitForNewMessages();
Message[] mailMessages = mailReceiver.receive();
if (logger.isDebugEnabled()) {
logger.debug("received " + mailMessages.length + " mail messages");
}
for (Message mailMessage : mailMessages) {
MimeMessage copied = new MimeMessage((MimeMessage) mailMessage);
sendMessage(MessageBuilder.withPayload(copied).build());
}
}
catch (MessagingException e) {
handleMailMessagingException(e);
return;
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
}
}
}

150
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ImapMailReceiver.java Executable file
View File

@@ -0,0 +1,150 @@
/*
* 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 javax.mail.Flags;
import javax.mail.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.event.MessageCountAdapter;
import javax.mail.event.MessageCountEvent;
import javax.mail.event.MessageCountListener;
import javax.mail.search.AndTerm;
import javax.mail.search.FlagTerm;
import javax.mail.search.SearchTerm;
import org.springframework.integration.mail.monitor.MailTransportUtils;
import org.springframework.util.Assert;
import com.sun.mail.imap.IMAPFolder;
/**
* @author Arjen Poutsma
* @author Mark Fisher
*/
public class ImapMailReceiver extends AbstractMailReceiver {
private volatile boolean shouldDeleteMessages = true;
private final MessageCountListener messageCountListener = new SimpleMessageCountListener();
public ImapMailReceiver(String url) {
super(url);
}
/**
* Specify whether mail messages should be deleted after retrieval.
* The default is <code>true</code>.
*/
public void setShouldDeleteMessages(boolean shouldDeleteMessages) {
this.shouldDeleteMessages = shouldDeleteMessages;
}
@Override
protected boolean shouldDeleteMessages() {
return this.shouldDeleteMessages;
}
/**
* This method is unique to the IMAP receiver and only works if IMAP IDLE
* is supported (see RFC 2177 for more detail).
*/
public void waitForNewMessages() throws MessagingException, InterruptedException {
this.openFolder();
Assert.state(this.getFolder() instanceof IMAPFolder,
"folder is not an instance of [" + IMAPFolder.class.getName() + "]");
IMAPFolder imapFolder = (IMAPFolder) this.getFolder();
if (imapFolder.hasNewMessages()) {
return;
}
imapFolder.addMessageCountListener(this.messageCountListener);
try {
imapFolder.idle();
}
finally {
imapFolder.removeMessageCountListener(this.messageCountListener);
}
}
/**
* Retrieves new messages from this receiver's folder. This implementation
* creates a {@link SearchTerm} that searches for all messages in the
* folder that are {@link javax.mail.Flags.Flag#RECENT RECENT}, not
* {@link javax.mail.Flags.Flag#ANSWERED ANSWERED}, and not
* {@link javax.mail.Flags.Flag#DELETED DELETED}. The search term is used
* to {@link Folder#search(SearchTerm) search} for new messages.
*
* @return the new messages
* @throws MessagingException in case of JavaMail errors
*/
@Override
protected Message[] searchForNewMessages() throws MessagingException {
Flags supportedFlags = this.getFolder().getPermanentFlags();
SearchTerm searchTerm = null;
if (supportedFlags != null) {
if (supportedFlags.contains(Flags.Flag.RECENT)) {
searchTerm = new FlagTerm(new Flags(Flags.Flag.RECENT), true);
}
if (supportedFlags.contains(Flags.Flag.ANSWERED)) {
FlagTerm answeredTerm = new FlagTerm(new Flags(Flags.Flag.ANSWERED), false);
if (searchTerm == null) {
searchTerm = answeredTerm;
}
else {
searchTerm = new AndTerm(searchTerm, answeredTerm);
}
}
if (supportedFlags.contains(Flags.Flag.DELETED)) {
FlagTerm deletedTerm = new FlagTerm(new Flags(Flags.Flag.DELETED), false);
if (searchTerm == null) {
searchTerm = deletedTerm;
}
else {
searchTerm = new AndTerm(searchTerm, deletedTerm);
}
}
}
Message[] results = searchTerm != null ? this.getFolder().search(searchTerm) : this.getFolder().getMessages();
if (results == null || results.length == 0) {
MailTransportUtils.closeFolder(this.getFolder());
}
return results;
}
/**
* Callback used for handling the event-driven idle response.
*/
private static class SimpleMessageCountListener extends MessageCountAdapter {
public void messagesAdded(MessageCountEvent event) {
Message[] messages = event.getMessages();
for (Message message : messages) {
try {
// this will return the flow to the idle call
message.getLineCount();
}
catch (MessagingException e) {
// ignored;
}
}
}
}
}

147
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ListeningMailSource.java
View File

@@ -1,147 +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 javax.mail.Message;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.endpoint.AbstractMessageProducingEndpoint;
import org.springframework.integration.mail.monitor.AsyncMonitoringStrategy;
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 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 ListeningMailSource extends AbstractMessageProducingEndpoint implements Lifecycle, DisposableBean {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile TaskExecutor taskExecutor;
private final MonitorRunnable monitorRunnable;
private volatile boolean monitorRunning = false;
public ListeningMailSource(FolderConnection folderConnection) {
Assert.notNull(folderConnection, "FolderConnection must not be null");
this.monitorRunnable = new MonitorRunnable(folderConnection);
}
// Lifecycle implementation
public boolean isRunning() {
return this.monitorRunning;
}
public void start() {
this.startMonitor();
if (logger.isInfoEnabled()) {
logger.info("started monitoring mailbox ["
+ this.monitorRunnable.folderConnection + "]");
}
}
public void stop() {
this.stopMonitor();
if (logger.isInfoEnabled()) {
logger.info("stopped monitoring mailbox ["
+ this.monitorRunnable.folderConnection + "]");
}
}
protected void startMonitor() {
synchronized (this.monitorRunnable) {
if (!this.monitorRunning) {
if (this.taskExecutor == null) {
if (logger.isInfoEnabled()) {
logger.info("No TaskExecutor has been provided, will use a ["
+ SimpleAsyncTaskExecutor.class + "] as the default.");
}
this.taskExecutor = new SimpleAsyncTaskExecutor();
}
this.taskExecutor.execute(this.monitorRunnable);
}
this.monitorRunning = true;
}
}
protected void stopMonitor() {
synchronized (this.monitorRunnable) {
if (this.monitorRunning) {
this.monitorRunnable.interrupt();
}
this.monitorRunning = false;
}
}
public void destroy() throws Exception {
this.stop();
}
private class MonitorRunnable implements Runnable {
private volatile Thread thread;
private final FolderConnection folderConnection;
private MonitorRunnable(FolderConnection folderConnection) {
Assert.notNull(folderConnection, "folderConnection must not be null");
this.folderConnection = folderConnection;
}
public synchronized void interrupt() {
if (this.thread != null) {
this.thread.interrupt();
}
else if (logger.isInfoEnabled()) {
logger.info("monitor is not running, cannot interrupt");
}
}
public void run() {
this.thread = Thread.currentThread();
while (!Thread.currentThread().isInterrupted()) {
if (!this.folderConnection.isRunning()) {
this.folderConnection.start();
}
Message[] mailMessages = this.folderConnection.receive();
for (Message mailMessage : mailMessages) {
ListeningMailSource.this.sendMessage(MessageBuilder.withPayload(mailMessage).build());
}
}
}
}
}

14
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/FolderConnection.java → org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailReceiver.java
View File

@@ -16,19 +16,13 @@
package org.springframework.integration.mail;
import javax.mail.Folder;
import javax.mail.Message;
import org.springframework.context.Lifecycle;
/**
* Encapsulates state for a restartable connection to a {@link Folder} and
* ensures thread safety.
* Strategy interface for receiving mail {@link javax.mail.Message Messages}.
*
* @author Jonas Partner
* @author Mark Fisher
*/
public interface FolderConnection extends Lifecycle {
public interface MailReceiver {
Message[] receive();
javax.mail.Message[] receive() throws javax.mail.MessagingException;
}

38
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/PollingMailSource.java → org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.mail;
import java.util.Arrays;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -24,50 +25,59 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.mail.monitor.MonitoringStrategy;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageSource;
import org.springframework.util.Assert;
/**
* {@link MessageSource} implementation that delegates to a
* {@link MonitoringStrategy} to poll a mailbox. Each poll of the mailbox may
* {@link MailReceiver} to poll a mailbox. Each poll of the mailbox may
* return more than one message which will then be stored in a queue.
*
* @author Jonas Partner
* @author Jonas Partner
* @author Mark Fisher
*/
public class PollingMailSource implements MessageSource<javax.mail.Message> {
public class MailReceivingMessageSource implements MessageSource<javax.mail.Message> {
private final Log logger = LogFactory.getLog(this.getClass());
private final FolderConnection folderConnection;
private final MailReceiver mailReceiver;
private final Queue<javax.mail.Message> mailQueue = new ConcurrentLinkedQueue<javax.mail.Message>();
public PollingMailSource(FolderConnection folderConnection) {
Assert.notNull(folderConnection, "folderConnection must not be null");
this.folderConnection = folderConnection;
public MailReceivingMessageSource(MailReceiver mailReceiver) {
Assert.notNull(mailReceiver, "mailReceiver must not be null");
this.mailReceiver = mailReceiver;
}
public MailReceivingMessageSource(String url) {
if (url.startsWith("imap")) {
this.mailReceiver = new Pop3MailReceiver(url);
}
else if (url.startsWith("pop3")) {
this.mailReceiver = new ImapMailReceiver(url);
}
else {
throw new UnsupportedOperationException("unsupported mail protocol '"
+ url.substring(0, url.indexOf(':')) + "'");
}
}
@SuppressWarnings("unchecked")
public Message<javax.mail.Message> receive() {
try {
javax.mail.Message mailMessage = this.mailQueue.poll();
if (mailMessage == null) {
javax.mail.Message[] messages = this.folderConnection.receive();
javax.mail.Message[] messages = this.mailReceiver.receive();
if (messages != null) {
for (javax.mail.Message message : messages) {
this.mailQueue.add(message);
}
this.mailQueue.addAll(Arrays.asList(messages));
}
mailMessage = this.mailQueue.poll();
}
if (mailMessage != null) {
if (logger.isDebugEnabled()) {
logger.debug("Received mail message [" + mailMessage + "]");
logger.debug("received mail message [" + mailMessage + "]");
}
return MessageBuilder.withPayload(mailMessage).build();
}

84
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/Pop3MailReceiver.java Executable file
View File

@@ -0,0 +1,84 @@
/*
* 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 javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.URLName;
import javax.mail.internet.MimeMessage;
import org.springframework.integration.mail.monitor.MailTransportUtils;
import org.springframework.util.Assert;
/**
* A {@link MailReceiver} implementation that polls a mail server using the
* POP3 protocol.
*
* @author Arjen Poutsma
* @author Mark Fisher
*/
public class Pop3MailReceiver extends AbstractMailReceiver {
public Pop3MailReceiver(String url) {
super(url);
Assert.isTrue(url.startsWith("pop3"), "url must start with 'pop3'");
}
public Pop3MailReceiver(String host, String username, String password) {
// -1 indicates default port
this(host, -1, username, password);
}
public Pop3MailReceiver(String host, int port, String username, String password) {
super(new URLName("pop3", host, port, "INBOX", username, password));
}
/**
* POP3 is unable to detect new Messages, so this always returns true.
*/
@Override
protected final boolean shouldDeleteMessages() {
return true;
}
@Override
protected Message[] searchForNewMessages() throws MessagingException {
int messageCount = this.getFolder().getMessageCount();
if (messageCount == 0) {
return new Message[0];
}
return this.getFolder().getMessages();
}
/**
* Deletes the given messages from this receiver's folder, and closes it to expunge deleted messages.
*
* @param messages the messages to delete
* @throws MessagingException in case of JavaMail errors
*/
@Override
protected void deleteMessages(Message[] messages) throws MessagingException {
super.deleteMessages(messages);
// expunge deleted mails, and make sure we've retrieved them before closing the folder
for (int i = 0; i < messages.length; i++) {
new MimeMessage((MimeMessage) messages[i]);
}
MailTransportUtils.closeFolder(this.getFolder(), true);
}
}

26
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/SubscribableImapIdleMailSourceParser.java → org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParser.java
View File

@@ -21,22 +21,20 @@ 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.mail.DefaultFolderConnection;
import org.springframework.integration.mail.ListeningMailSource;
import org.springframework.integration.mail.monitor.ImapIdleMonitoringStrategy;
import org.springframework.integration.mail.ImapIdleChannelAdapter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;imap-idle-mail-source&gt; element in the 'mail' namespace.
* Parser for the &lt;imap-idle-channel-adapter&gt; element in the 'mail' namespace.
*
* @author Jonas Partner
* @author Mark Fisher
*/
public class SubscribableImapIdleMailSourceParser extends AbstractSingleBeanDefinitionParser {
public class ImapIdleChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
protected Class<?> getBeanClass(Element element) {
return ListeningMailSource.class;
return ImapIdleChannelAdapter.class;
}
protected boolean shouldGenerateId() {
@@ -51,24 +49,20 @@ public class SubscribableImapIdleMailSourceParser extends AbstractSingleBeanDefi
String channel = element.getAttribute("channel");
String uri = element.getAttribute("store-uri");
String taskExecutorRef = element.getAttribute("task-executor");
String propertiesRef = element.getAttribute("javaMailProperties");
String propertiesRef = element.getAttribute("java-mail-properties");
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());
// set polling false
folderConnectionBuilder.addConstructorArgValue(false);
"store-uri must start with 'imap' for the imap idle channel adapter");
builder.addConstructorArgValue(uri);
if (StringUtils.hasText(propertiesRef)) {
folderConnectionBuilder.addPropertyReference("javaMailProperties", propertiesRef);
builder.addPropertyReference("javaMailProperties", propertiesRef);
}
builder.addConstructorArgValue(folderConnectionBuilder.getBeanDefinition());
if (StringUtils.hasLength(taskExecutorRef)) {
builder.addPropertyReference("taskExecutor", taskExecutorRef);
}
builder.addPropertyValue("shouldDeleteMessages",
!"false".equals(element.getAttribute("should-delete-messages")));
builder.addPropertyReference("outputChannel", channel);
}

61
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/MailInboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* 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.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.mail.ImapMailReceiver;
import org.springframework.integration.mail.MailReceiver;
import org.springframework.integration.mail.MailReceivingMessageSource;
import org.springframework.integration.mail.Pop3MailReceiver;
import org.springframework.util.Assert;
/**
* Parser for the &lt;inbound-channel-adapter&gt; element of Spring
* Integration's 'mail' namespace.
*
* @author Jonas Partner
* @author Mark Fisher
*/
public class MailInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected String parseSource(Element element, ParserContext parserContext) {
String uri = element.getAttribute("store-uri");
//String propertiesRef = element.getAttribute("javaMailProperties");
Assert.hasText(uri, "the 'store-uri' attribute is required");
boolean isPop3 = uri.toLowerCase().startsWith("pop3");
boolean isImap = uri.toLowerCase().startsWith("imap");
Assert.isTrue(isPop3 || isImap, "the 'store-uri' must begin with 'pop3' or 'imap'");
MailReceiver mailReceiver = isPop3 ? new Pop3MailReceiver(uri) : new ImapMailReceiver(uri);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MailReceivingMessageSource.class);
/*
if (StringUtils.hasText(propertiesRef)) {
folderConnectionBuilder.addPropertyReference("javaMailProperties",
propertiesRef);
}
*/
builder.addConstructorArgValue(mailReceiver);
return BeanDefinitionReaderUtils.registerWithGeneratedName(
builder.getBeanDefinition(), parserContext.getRegistry());
}
}

4
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/MailNamespaceHandler.java
View File

@@ -29,8 +29,8 @@ public class MailNamespaceHandler extends NamespaceHandlerSupport {
public void init() {
this.registerBeanDefinitionParser("outbound-channel-adapter", new MailOutboundChannelAdapterParser());
this.registerBeanDefinitionParser("polling-mail-source", new PollingMailSourceParser());
this.registerBeanDefinitionParser("imap-idle-mail-source", new SubscribableImapIdleMailSourceParser());
this.registerBeanDefinitionParser("inbound-channel-adapter", new MailInboundChannelAdapterParser());
this.registerBeanDefinitionParser("imap-idle-channel-adapter", new ImapIdleChannelAdapterParser());
this.registerBeanDefinitionParser("header-enricher", new SimpleHeaderEnricherParser(MailHeaders.PREFIX));
this.registerBeanDefinitionParser("mail-to-string-transformer", new MailToStringTransformerParser());
}

85
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/PollingMailSourceParser.java
View File

@@ -1,85 +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.config;
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.mail.DefaultFolderConnection;
import org.springframework.integration.mail.PollingMailSource;
import org.springframework.integration.mail.monitor.MonitoringStrategy;
import org.springframework.integration.mail.monitor.PollingMonitoringStrategy;
import org.springframework.integration.mail.monitor.Pop3PollingMonitoringStrategy;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Jonas Partner
*/
public class PollingMailSourceParser extends AbstractSingleBeanDefinitionParser {
protected Class<?> getBeanClass(Element element) {
return PollingMailSource.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String mailConvertorRef = element.getAttribute("convertor");
String uri = element.getAttribute("store-uri");
String propertiesRef = element.getAttribute("javaMailProperties");
Assert.hasText(uri, "the 'store-uri' attribute is required");
BeanDefinitionBuilder folderConnectionBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultFolderConnection.class);
MonitoringStrategy monitoringStrategy = null;
if (uri.toLowerCase().startsWith("pop3")) {
monitoringStrategy = new Pop3PollingMonitoringStrategy();
}
else if (uri.toLowerCase().startsWith("imap")) {
monitoringStrategy = new PollingMonitoringStrategy();
}
else {
throw new IllegalArgumentException(
"unable to determine monitoring strategy for store-uri [" + uri + "]");
}
folderConnectionBuilder.addConstructorArgValue(uri);
folderConnectionBuilder.addConstructorArgValue(monitoringStrategy);
// set polling true
folderConnectionBuilder.addConstructorArgValue(true);
if (StringUtils.hasText(propertiesRef)) {
folderConnectionBuilder.addPropertyReference("javaMailProperties",
propertiesRef);
}
String folderConnectionName = parserContext.getReaderContext()
.registerWithGeneratedName(folderConnectionBuilder.getBeanDefinition());
builder.addDependsOn(folderConnectionName);
builder.addConstructorArgValue(folderConnectionBuilder.getBeanDefinition());
if (StringUtils.hasText(mailConvertorRef)) {
builder.addPropertyReference("convertor", mailConvertorRef);
}
}
}

24
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/config/spring-integration-mail-1.0.xsd
View File

@@ -3,12 +3,14 @@
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/mail"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -34,29 +36,35 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="polling-mail-source">
<xsd:element name="inbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a polling mail source.
Defines an inbound Channel Adapter that polls a mailbox for mail messages.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string"/>
<xsd:attribute name="store-uri" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="imap-idle-mail-source">
<xsd:element name="imap-idle-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an IMAP mail source.
Defines an IMAP IDLE channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="channel" type="xsd:string"/>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
<xsd:attribute name="store-uri" type="xsd:string" use="required"/>
<xsd:attribute name="task-executor" type="xsd:string" use="optional"/>
<xsd:attribute name="java-mail-properties" type="xsd:string"/>
<xsd:attribute name="task-executor" type="xsd:string"/>
<xsd:attribute name="should-delete-messages" type="xsd:string"/>
</xsd:complexType>
</xsd:element>

169
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/AbstractMonitoringStrategy.java
View File

@@ -1,169 +0,0 @@
/*
* Copyright 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.FetchProfile;
import javax.mail.Flags;
import javax.mail.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.search.AndTerm;
import javax.mail.search.FlagTerm;
import javax.mail.search.SearchTerm;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Abstract base class for the {@link MonitoringStrategy} interface. Exposes a {@link #setDeleteMessages(boolean)
* deleteMessages} property, and includes a basic workflow for message monitoring.
*
* @author Arjen Poutsma
*/
public abstract class AbstractMonitoringStrategy implements MonitoringStrategy {
/** Logger available to subclasses. */
protected final Log logger = LogFactory.getLog(getClass());
private boolean deleteMessages = true;
private int maxMessagesPerReceive = -1;
/**
* Sets whether messages should be marked as {@link javax.mail.Flags.Flag#DELETED DELETED} after they have been
* read. Default is <code>true</code>.
*/
public void setDeleteMessages(boolean deleteMessages) {
this.deleteMessages = deleteMessages;
}
public int getFolderOpenMode() {
return deleteMessages ? Folder.READ_WRITE : Folder.READ_ONLY;
}
public void setMaxMessagePerDownload(int maxMessagesPerReceive){
this.maxMessagesPerReceive = maxMessagesPerReceive;
}
/**
* Monitors the given folder, and returns any new messages when they arrive. This implementation calls {@link
* #waitForNewMessages(Folder)}, then searches for new messages using {@link #searchForNewMessages(Folder)}, fetches
* the messages using {@link #fetchMessages(Folder, Message[])}, and finally {@link #setDeleteMessages(boolean)
* deletes} the messages, if {@link #setDeleteMessages(boolean) deleteMessages} is <code>true</code>.
*
* @param folder the folder to monitor
* @return the new messages
* @throws MessagingException in case of JavaMail errors
* @throws InterruptedException when a thread is interrupted
*/
public final Message[] receive(Folder folder) throws MessagingException, InterruptedException {
logger.info("Receiving for folder" + folder.getFullName());
if(!folder.isOpen()){
folder.open(getFolderOpenMode());
}
folder.getMessageCount();
Message[] messages = searchForNewMessages(folder);
if (logger.isDebugEnabled()) {
logger.debug("Found " + messages.length + " new messages");
}
if(maxMessagesPerReceive > 0 && messages.length > maxMessagesPerReceive){
Message[] reducedMessages = new Message[maxMessagesPerReceive];
System.arraycopy(messages, 0, reducedMessages, 0, maxMessagesPerReceive);
messages = reducedMessages;
}
if (messages.length > 0) {
fetchMessages(folder, messages);
}
if (deleteMessages) {
deleteMessages(folder, messages);
}
return messages;
}
/**
* Retrieves new messages from the given folder. This implementation creates a {@link SearchTerm} that searches for
* all messages in the folder that are {@link javax.mail.Flags.Flag#RECENT RECENT}, not {@link
* javax.mail.Flags.Flag#ANSWERED ANSWERED}, and not {@link javax.mail.Flags.Flag#DELETED DELETED}. The search term
* is used to {@link Folder#search(SearchTerm) search} for new messages.
*
* @param folder the folder to retrieve new messages from
* @return the new messages
* @throws MessagingException in case of JavaMail errors
*/
protected Message[] searchForNewMessages(Folder folder) throws MessagingException {
if (!folder.isOpen()) {
return new Message[0];
}
Flags supportedFlags = folder.getPermanentFlags();
SearchTerm searchTerm = null;
if (supportedFlags != null) {
if (supportedFlags.contains(Flags.Flag.RECENT)) {
searchTerm = new FlagTerm(new Flags(Flags.Flag.RECENT), true);
}
if (supportedFlags.contains(Flags.Flag.ANSWERED)) {
FlagTerm answeredTerm = new FlagTerm(new Flags(Flags.Flag.ANSWERED), false);
if (searchTerm == null) {
searchTerm = answeredTerm;
}
else {
searchTerm = new AndTerm(searchTerm, answeredTerm);
}
}
if (supportedFlags.contains(Flags.Flag.DELETED)) {
FlagTerm deletedTerm = new FlagTerm(new Flags(Flags.Flag.DELETED), false);
if (searchTerm == null) {
searchTerm = deletedTerm;
}
else {
searchTerm = new AndTerm(searchTerm, deletedTerm);
}
}
}
return searchTerm != null ? folder.search(searchTerm) : folder.getMessages();
}
/**
* Fetches the specified messages from the specified folder. Default implementation {@link Folder#fetch(Message[],
* FetchProfile) fetches} every {@link javax.mail.FetchProfile.Item}.
*
* @param folder the folder to fetch messages from
* @param messages the messages to fetch
* @throws MessagingException in case of JavMail errors
*/
protected void fetchMessages(Folder folder, Message[] messages) throws MessagingException {
FetchProfile contentsProfile = new FetchProfile();
contentsProfile.add(FetchProfile.Item.ENVELOPE);
contentsProfile.add(FetchProfile.Item.CONTENT_INFO);
contentsProfile.add(FetchProfile.Item.FLAGS);
folder.fetch(messages, contentsProfile);
}
/**
* Deletes the given messages from the given folder. Only invoked when {@link #setDeleteMessages(boolean)} is
* <code>true</code>.
*
* @param folder the folder to delete messages from
* @param messages the messages to delete
* @throws MessagingException in case of JavaMail errors
*/
protected void deleteMessages(Folder folder, Message[] messages) throws MessagingException {
for (int i = 0; i < messages.length; i++) {
messages[i].setFlag(Flags.Flag.DELETED, true);
}
}
}

26
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/AsyncMonitoringStrategy.java
View File

@@ -1,26 +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.Folder;
import javax.mail.MessagingException;
public interface AsyncMonitoringStrategy {
public abstract void waitForNewMessages(Folder folder)
throws MessagingException, InterruptedException;
}

81
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/ImapIdleMonitoringStrategy.java
View File

@@ -1,81 +0,0 @@
/*
* Copyright 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.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.event.MessageCountAdapter;
import javax.mail.event.MessageCountEvent;
import javax.mail.event.MessageCountListener;
import org.springframework.util.Assert;
import com.sun.mail.imap.IMAPFolder;
/**
* Implementation of the {@link MonitoringStrategy} interface that uses the IMAP IDLE command for asynchronous message
* detection.
* <p/>
* <b>Note</b> that this implementation is only suitable for use with IMAP servers which support the IDLE command.
* Additionally, this strategy requires JavaMail version 1.4.1.
*
* @author Arjen Poutsma
*/
public class ImapIdleMonitoringStrategy extends AbstractMonitoringStrategy implements AsyncMonitoringStrategy {
private MessageCountListener messageCountListener;
/* (non-Javadoc)
* @see org.springframework.integration.adapter.mail.monitor.AynchronouseMonitoringStrategy#waitForNewMessages(javax.mail.Folder)
*/
public void waitForNewMessages(Folder folder) throws MessagingException, InterruptedException {
Assert.isInstanceOf(IMAPFolder.class, folder);
IMAPFolder imapFolder = (IMAPFolder) folder;
//retrieve unseen messages before we enter the blocking idle call
if (searchForNewMessages(folder).length > 0) {
return;
}
if (messageCountListener == null) {
createMessageCountListener();
}
folder.addMessageCountListener(messageCountListener);
try {
imapFolder.idle();
}
finally {
folder.removeMessageCountListener(messageCountListener);
}
}
private void createMessageCountListener() {
messageCountListener = new MessageCountAdapter() {
public void messagesAdded(MessageCountEvent e) {
Message[] messages = e.getMessages();
for (int i = 0; i < messages.length; i++) {
try {
// this will return the flow to the idle call, above
messages[i].getLineCount();
}
catch (MessagingException ex) {
// ignore
}
}
}
};
}
}

36
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/MailTransportConstants.java
View File

@@ -1,36 +0,0 @@
/*
* Copyright 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;
/**
* Declares Mail-specific transport constants.
*
* @author Arjen Poutsma
*/
public interface MailTransportConstants {
/**
* The "mail" URI scheme.
*/
String MAIL_URI_SCHEME = "mailto";
/**
* The "In-Reply-To" header.
*/
String HEADER_IN_REPLY_TO = "In-Reply-To";
}

18
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/MailTransportUtils.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.mail.monitor;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -167,21 +166,4 @@ public abstract class MailTransportUtils {
return tempURL.toString();
}
/**
* Converts the given internet address into a <code>mailto</code> URI.
*
* @param to the To: address
* @param subject the subject, may be <code>null</code>
* @return a mailto URI
*/
public static URI toUri(InternetAddress to, String subject) throws URISyntaxException {
if (StringUtils.hasLength(subject)) {
return new URI(MailTransportConstants.MAIL_URI_SCHEME, to.getAddress() + "?subject=" + subject, null);
}
else {
return new URI(MailTransportConstants.MAIL_URI_SCHEME, to.getAddress(), null);
}
}
}

47
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/MonitoringStrategy.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright 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.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
/**
* Defines the contract for objects that monitor a given folder for new messages. Allows for multiple implementation
* strategies, including polling, or event-driven techniques such as IMAP's <code>IDLE</code> command.
*
* @author Arjen Poutsma
*/
public interface MonitoringStrategy {
/**
* Monitors the given folder, and returns any new messages when they arrive.
*
* @param folder the folder in which to look for new messages
* @return the new messages
* @throws MessagingException in case of JavaMail errors
* @throws InterruptedException if a thread is interrupted
*/
Message[] receive(Folder folder) throws MessagingException, InterruptedException;
/**
* Returns the folder open mode to be used by this strategy. Can be either {@link Folder#READ_ONLY} or {@link
* Folder#READ_WRITE}.
*/
int getFolderOpenMode();
}

33
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/PollingMonitoringStrategy.java
View File

@@ -1,33 +0,0 @@
/*
* Copyright 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;
/**
* Implementation of the {@link MonitoringStrategy} interface that uses a simple polling mechanism. Defines a {@link
* #setPollingInterval(long) polling interval} property which defines the interval in between message polls.
* <p/>
* <b>Note</b> that this implementation is not suitable for use with POP3 servers. Use the {@link
* Pop3PollingMonitoringStrategy} instead.
*
* @author Arjen Poutsma
*/
public class PollingMonitoringStrategy extends AbstractMonitoringStrategy {
}

66
org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/monitor/Pop3PollingMonitoringStrategy.java
View File

@@ -1,66 +0,0 @@
/*
* Copyright 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.Folder;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;
/**
* Implementation of the {@link MonitoringStrategy} interface that uses a simple polling mechanism suitable for POP3
* servers. Since POP3 does not have a native mechanism to determine which messages are "new", this implementation
* simply retrieves all messages in the {@link Folder}, and delete them afterwards. All messages in the POP3 mailbox are
* therefore, by definition, new.
* <p/>
* Setting the {@link #setDeleteMessages(boolean) deleteMessages} property is therefore ignored: messages are always
* deleted.
*
* @author Arjen Poutsma
*/
public class Pop3PollingMonitoringStrategy extends PollingMonitoringStrategy {
public Pop3PollingMonitoringStrategy() {
super.setDeleteMessages(true);
}
public void setDeleteMessages(boolean deleteMessages) {
}
/**
* Simply returns {@link Folder#getMessages()}.
*/
protected Message[] searchForNewMessages(Folder folder) throws MessagingException {
return folder.getMessages();
}
/**
* Deletes the given messages from the given folder, and closes it to expunge deleted messages.
*
* @param folder the folder to delete messages from
* @param messages the messages to delete
* @throws MessagingException in case of JavaMail errors
*/
protected void deleteMessages(Folder folder, Message[] messages) throws MessagingException {
super.deleteMessages(folder, messages);
// expunge deleted mails, and make sure we've retrieved them before closing the folder
for (int i = 0; i < messages.length; i++) {
new MimeMessage((MimeMessage) messages[i]);
}
MailTransportUtils.closeFolder(folder, true);
}
}

25
org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/PollingMessageSourceTests.java → org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/MailReceivingMessageSourceTests.java
View File

@@ -28,31 +28,32 @@ import org.junit.Test;
/**
* @author Jonas Partner
* @author Mark Fisher
*/
public class PollingMessageSourceTests {
public class MailReceivingMessageSourceTests {
@Test
public void testPolling() {
StubFolderConnection folderConnection = new StubFolderConnection();
StubMailReceiver mailReceiver = new StubMailReceiver();
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 });
mailReceiver.messages.add(new javax.mail.Message[] { message1 });
mailReceiver.messages.add(new javax.mail.Message[] { message2, message3 });
mailReceiver.messages.add(new javax.mail.Message[] { message4 });
PollingMailSource pollingMailSource = new PollingMailSource(folderConnection);
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());
MailReceivingMessageSource source = new MailReceivingMessageSource(mailReceiver);
assertEquals("Wrong message for number 1", message1, source.receive().getPayload());
assertEquals("Wrong message for number 2", message2, source.receive().getPayload());
assertEquals("Wrong message for number 3", message3, source.receive().getPayload());
assertEquals("Wrong message for number 4", message4, source.receive().getPayload());
assertNull("Expected null after exhausting all messages", source.receive());
}
private static class StubFolderConnection implements FolderConnection {
private static class StubMailReceiver implements MailReceiver {
private final ConcurrentLinkedQueue<javax.mail.Message[]> messages = new ConcurrentLinkedQueue<javax.mail.Message[]>();

80
org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/SubscribableMailSourceTests.java
View File

@@ -1,80 +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 java.util.concurrent.ConcurrentLinkedQueue;
import javax.mail.internet.MimeMessage;
import org.easymock.classextension.EasyMock;
import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
/**
* @author Jonas Partner
*/
public class SubscribableMailSourceTests {
@Test
public void testReceive() throws Exception {
javax.mail.Message message = EasyMock.createMock(MimeMessage.class);
StubFolderConnection folderConnection = new StubFolderConnection(message);
QueueChannel channel = new QueueChannel();
ListeningMailSource mailSource = new ListeningMailSource(folderConnection);
mailSource.setOutputChannel(channel);
mailSource.start();
Message<?> result = channel.receive(1000);
mailSource.stop();
assertNotNull(result);
assertEquals("Wrong payload", message, result.getPayload());
mailSource.stop();
}
private static class StubFolderConnection implements FolderConnection {
private final ConcurrentLinkedQueue<javax.mail.Message> messages = new ConcurrentLinkedQueue<javax.mail.Message>();
public StubFolderConnection(javax.mail.Message message) {
messages.add(message);
}
public javax.mail.Message[] receive() {
javax.mail.Message msg = messages.poll();
if (msg == null) {
return new javax.mail.Message[] {};
}
return new javax.mail.Message[] { msg };
}
public boolean isRunning() {
return false;
}
public void start() {
}
public void stop() {
}
}
}

12
org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/PollingMailSourceParserTests.java
View File

@@ -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,19 +13,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.mail.config;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.mail.PollingMailSource;
/**
* @author Jonas Partner
*/
public class PollingMailSourceParserTests {
@Test
public void testPop3(){
ApplicationContext context = new ClassPathXmlApplicationContext("pollingMailSourceParserTests.xml", PollingMailSourceParserTests.class);
PollingMailSource mailSource = (PollingMailSource)context.getBean("pollingPop3");
context.getBean("pollingPop3");
}
}

3
org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/mailOutboundChannelAdapterParserTests.xml
View File

@@ -2,8 +2,11 @@
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:mail="http://www.springframework.org/schema/integration/mail"
xmlns:integration="http://wwww.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
http://www.springframework.org/schema/integration/mail
http://www.springframework.org/schema/integration/mail/spring-integration-mail-1.0.xsd">

5
org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/pollingMailSourceParserTests.xml
View File

@@ -2,11 +2,14 @@
<beans:beans xmlns="http://www.springframework.org/schema/integration/mail"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:integration="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
http://www.springframework.org/schema/integration/mail
http://www.springframework.org/schema/integration/mail/spring-integration-mail-1.0.xsd">
<polling-mail-source id="pollingPop3" store-uri="pop3://mailtest:mailtest@ubuntuservervm/INBOX" />
<inbound-channel-adapter id="pollingPop3" store-uri="pop3://mailtest:mailtest@ubuntuservervm/INBOX" />
</beans:beans>