diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java index b811ecafd6..a92c091164 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java @@ -57,8 +57,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { private volatile ScheduledFuture receivingTask; private volatile ScheduledFuture pingTask; - private volatile ResubmittingTask task; - private volatile long connectionPingInterval = 10000; public ImapIdleChannelAdapter(ImapMailReceiver mailReceiver) { @@ -73,7 +71,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { public void setShouldReconnectAutomatically(boolean shouldReconnectAutomatically) { this.shouldReconnectAutomatically = shouldReconnectAutomatically; } - + public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = taskExecutor; } @@ -93,12 +91,25 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { @Override // guarded by super#lifecycleLock protected void doStart() { - TaskScheduler scheduler = this.getTaskScheduler(); + final TaskScheduler scheduler = this.getTaskScheduler(); Assert.notNull(scheduler, "'taskScheduler' must not be null" ); - this.task = new ResubmittingTask(this.idleTask, scheduler, reconnectDelay); - task.setTaskExecutor(taskExecutor); - receivingTask = scheduler.schedule(task, new Date()); + receivingTask = scheduler.scheduleAtFixedRate(new Runnable(){ + + public void run() { + try { + idleTask.run(); + if (logger.isDebugEnabled()){ + logger.debug("Task completed successfully. Re-scheduling it again right away"); + } + } + catch (IllegalStateException e) { //run again after a delay + logger.warn("Failed to execute IDLE task. Will atempt to resubmit in " + reconnectDelay + " milliseconds", e); + scheduler.scheduleAtFixedRate(this, new Date(System.currentTimeMillis() + reconnectDelay), 1); + } + } + + }, 1); pingTask = scheduler.scheduleAtFixedRate(new Runnable() { public void run() { @@ -116,7 +127,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { @Override // guarded by super#lifecycleLock protected void doStop() { - this.task.requestStop(); receivingTask.cancel(true); pingTask.cancel(true); try { @@ -135,14 +145,23 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { logger.debug("waiting for mail"); } mailReceiver.waitForNewMessages(); - if (!task.isStopRequested()){ + if (mailReceiver.getFolder().isOpen()){ 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()); + final MimeMessage copied = new MimeMessage((MimeMessage) mailMessage); + if (taskExecutor != null){ + taskExecutor.execute(new Runnable() { + public void run() { + sendMessage(MessageBuilder.withPayload(copied).build()); + } + }); + } + else { + sendMessage(MessageBuilder.withPayload(copied).build()); + } } } } catch (MessagingException e) { diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ResubmittingTask.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ResubmittingTask.java deleted file mode 100644 index e979d711b9..0000000000 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ResubmittingTask.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * Copyright 2002-2011 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.Date; -import java.util.concurrent.Executor; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.scheduling.TaskScheduler; - -/** - * @author Mark Fisher - * @author Oleg Zhurakousky - * @since 2.0.2 - * - * Will run the provided task right away resubmitting it after each successful execution - * or after receiving a {@link IllegalStateException}. If resubmission is after exception then - * the delay will be applied before resubmission. This is useful when the underlying task deals - * with reconnection logic - * Currently only used to manage IDLE task of ImapIdleChannelAdapter - */ -class ResubmittingTask implements Runnable { - private static final Log logger = LogFactory.getLog(ResubmittingTask.class); - private final Runnable targetTask; - private final TaskScheduler scheduler; - private final long delay; - private Executor taskExecutor = new SimpleAsyncTaskExecutor(); - - private volatile boolean stopRequested = false; - - public ResubmittingTask(Runnable targetTask, TaskScheduler scheduler, long delay) { - this.targetTask = targetTask; - this.scheduler = scheduler; - this.delay = delay; - } - - public void setTaskExecutor(Executor taskExecutor) { - this.taskExecutor = (taskExecutor != null) ? taskExecutor : new SimpleAsyncTaskExecutor(); - } - - public void run() { - taskExecutor.execute(new Runnable() { - public void run() { - ResubmittingTask.this.invokeTask(); - } - }); - } - - protected void requestStop(){ - this.stopRequested = true; - } - - protected boolean isStopRequested(){ - return this.stopRequested; - } - - private void invokeTask(){ - try { - targetTask.run(); - if (!this.stopRequested){ - if (logger.isDebugEnabled()){ - logger.debug("Task completed successfully. Re-scheduling it again right away"); - } - scheduler.schedule(this, new Date()); - } - else { - if (logger.isDebugEnabled()){ - logger.debug("IDLE Task is stopped"); - } - } - - } - catch (IllegalStateException e) { //run again after a delay - logger.warn("Failed to execute IDLE task. Will atempt to resubmit in " + delay + " milliseconds", e); - scheduler.schedule(this, new Date(System.currentTimeMillis() + delay)); - } - } -} diff --git a/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.0.xsd b/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.0.xsd index e1aa58f8fd..2d8b0786ab 100644 --- a/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.0.xsd +++ b/spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-2.0.xsd @@ -107,7 +107,7 @@ - Reference to a TaskExecutor for sending inbound Mail to a MessageChannel asynchronously. + Reference to a TaskExecutor which will be used to send inbound Mail to a MessageChannel. diff --git a/spring-integration-mail/src/test/java/log4j.properties b/spring-integration-mail/src/test/java/log4j.properties index c82ed7cf3a..14a9ecfa47 100644 --- a/spring-integration-mail/src/test/java/log4j.properties +++ b/spring-integration-mail/src/test/java/log4j.properties @@ -1,4 +1,4 @@ -log4j.rootCategory=WARN, stdout +log4j.rootCategory=DEBUG, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout