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

GH-8577: Revise ImapIdleChannelAdapter logic (#8588)

Browse Source 
* GH-8577: Revise `ImapIdleChannelAdapter` logic

Fixes https://github.com/spring-projects/spring-integration/issues/8577

When we process mail messages in async manner, it is possible that we end up
in a race condition situation where the next idle cycle closes the folder.

It is possible to reopen the folder, but feels better to block the current idle
cycle until we are done with the message and therefore keep folder opened.

* Deprecate `ImapIdleChannelAdapter.sendingTaskExecutor` in favor of an `ExecutorChannel`
as an output for this channel adapter or similar async hand-off downstream.
* Make use of `shouldReconnectAutomatically` as it is advertised for this channel adapter
* Optimize the proxy creation for message sending task

* * Remove `ImapIdleChannelAdapter.sendingTaskExecutor`

* Fix language in docs

Co-authored-by: Gary Russell <grussell@vmware.com>

---------

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2023-03-29 10:15:47 -04:00
committed by GitHub
parent 22d47e72e9
commit 4fdbdf180e
9 changed files with 125 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -19,14 +19,12 @@ package org.springframework.integration.mail;
import java.io.Serial;
import java.time.Instant;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import jakarta.mail.Folder;
import jakarta.mail.Message;
import jakarta.mail.MessagingException;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
@@ -38,6 +36,7 @@ import org.springframework.integration.mail.event.MailIntegrationEvent;
import org.springframework.integration.transaction.IntegrationResourceHolder;
import org.springframework.integration.transaction.IntegrationResourceHolderSynchronization;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.messaging.MessagingException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
@@ -78,12 +77,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
private boolean shouldReconnectAutomatically = true;
private Executor sendingTaskExecutor = Executors.newFixedThreadPool(1);
private boolean sendingTaskExecutorSet;
private List<Advice> adviceChain;
private Consumer<Object> messageSender;
private long reconnectDelay = DEFAULT_RECONNECT_DELAY; // milliseconds
private volatile ScheduledFuture<?> receivingTask;
@@ -103,21 +100,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
this.adviceChain = adviceChain;
}
/**
* Specify an {@link Executor} used to send messages received by the
* adapter.
* @param sendingTaskExecutor the sendingTaskExecutor to set
*/
public void setSendingTaskExecutor(Executor sendingTaskExecutor) {
Assert.notNull(sendingTaskExecutor, "'sendingTaskExecutor' must not be null");
this.sendingTaskExecutor = sendingTaskExecutor;
this.sendingTaskExecutorSet = true;
}
/**
* Specify whether the IDLE task should reconnect automatically after
* catching a {@link jakarta.mail.FolderClosedException} while waiting for messages. The
* default value is <code>true</code>.
* catching a {@link jakarta.mail.MessagingException} while waiting for messages. The
* default value is true.
* @param shouldReconnectAutomatically true to reconnect.
*/
public void setShouldReconnectAutomatically(boolean shouldReconnectAutomatically) {
@@ -148,6 +134,26 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
@SuppressWarnings("unchecked")
protected void onInit() {
super.onInit();
Consumer<?> messageSenderToUse = new MessageSender();
if (!CollectionUtils.isEmpty(this.adviceChain)) {
ProxyFactory proxyFactory = new ProxyFactory(messageSenderToUse);
this.adviceChain.forEach(proxyFactory::addAdvice);
for (Advice advice : this.adviceChain) {
proxyFactory.addAdvice(advice);
}
messageSenderToUse = (Consumer<?>) proxyFactory.getProxy(this.classLoader);
}
this.messageSender = (Consumer<Object>) messageSenderToUse;
}
/*
* Lifecycle implementation
*/
@@ -162,7 +168,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
@Override
// guarded by super#lifecycleLock
protected void doStop() {
this.receivingTask.cancel(true);
if (this.receivingTask != null) {
this.receivingTask.cancel(true);
this.receivingTask = null;
}
this.mailReceiver.cancelPing();
}
@@ -170,58 +179,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
public void destroy() {
super.destroy();
this.mailReceiver.destroy();
// If we're running with the default executor, shut it down.
if (!this.sendingTaskExecutorSet && this.sendingTaskExecutor != null) {
((ExecutorService) this.sendingTaskExecutor).shutdown();
}
}
private Runnable createMessageSendingTask(Object mailMessage) {
Runnable sendingTask = prepareSendingTask(mailMessage);
// wrap in the TX proxy if necessary
if (!CollectionUtils.isEmpty(this.adviceChain)) {
ProxyFactory proxyFactory = new ProxyFactory(sendingTask);
if (!CollectionUtils.isEmpty(this.adviceChain)) {
for (Advice advice : this.adviceChain) {
proxyFactory.addAdvice(advice);
}
}
sendingTask = (Runnable) proxyFactory.getProxy(this.classLoader);
}
return sendingTask;
}
private Runnable prepareSendingTask(Object mailMessage) {
return () -> {
@SuppressWarnings("unchecked")
org.springframework.messaging.Message<?> message =
mailMessage instanceof Message
? getMessageBuilderFactory().withPayload(mailMessage).build()
: (org.springframework.messaging.Message<Object>) mailMessage;
if (TransactionSynchronizationManager.isActualTransactionActive()
&& this.transactionSynchronizationFactory != null) {
TransactionSynchronization synchronization = this.transactionSynchronizationFactory.create(this);
if (synchronization != null) {
TransactionSynchronizationManager.registerSynchronization(synchronization);
if (synchronization instanceof IntegrationResourceHolderSynchronization
&& !TransactionSynchronizationManager.hasResource(this)) {
TransactionSynchronizationManager.bindResource(this,
((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder());
}
Object resourceHolder = TransactionSynchronizationManager.getResource(this);
if (resourceHolder instanceof IntegrationResourceHolder) {
((IntegrationResourceHolder) resourceHolder).setMessage(message);
}
}
}
sendMessage(message);
};
}
private void publishException(Exception ex) {
@@ -233,6 +190,42 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
}
}
private class MessageSender implements Consumer<Object> {
MessageSender() {
}
@Override
public void accept(Object mailMessage) {
org.springframework.messaging.Message<?> messageToSend =
mailMessage instanceof Message
? getMessageBuilderFactory().withPayload(mailMessage).build()
: (org.springframework.messaging.Message<?>) mailMessage;
if (TransactionSynchronizationManager.isActualTransactionActive()
&& ImapIdleChannelAdapter.this.transactionSynchronizationFactory != null) {
TransactionSynchronization synchronization =
ImapIdleChannelAdapter.this.transactionSynchronizationFactory.create(this);
if (synchronization != null) {
TransactionSynchronizationManager.registerSynchronization(synchronization);
if (synchronization instanceof IntegrationResourceHolderSynchronization integrationSync
&& !TransactionSynchronizationManager.hasResource(this)) {
TransactionSynchronizationManager.bindResource(this, integrationSync.getResourceHolder());
}
Object resourceHolder = TransactionSynchronizationManager.getResource(this);
if (resourceHolder instanceof IntegrationResourceHolder integrationResourceHolder) {
integrationResourceHolder.setMessage(messageToSend);
}
}
}
sendMessage(messageToSend);
}
}
private class ReceivingTask implements Runnable {
@@ -246,10 +239,23 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
ImapIdleChannelAdapter.this.idleTask.run();
logger.debug("Task completed successfully. Re-scheduling it again right away.");
}
catch (Exception ex) { //run again after a delay
logger.warn(ex, () -> "Failed to execute IDLE task. Will attempt to resubmit in "
+ ImapIdleChannelAdapter.this.reconnectDelay + " milliseconds.");
ImapIdleChannelAdapter.this.receivingTaskTrigger.delayNextExecution();
catch (Exception ex) {
if (ImapIdleChannelAdapter.this.shouldReconnectAutomatically
&& ex.getCause() instanceof jakarta.mail.MessagingException messagingException) {
//run again after a delay
logger.info(messagingException,
() -> "Failed to execute IDLE task. Will attempt to resubmit in "
+ ImapIdleChannelAdapter.this.reconnectDelay + " milliseconds.");
ImapIdleChannelAdapter.this.receivingTaskTrigger.delayNextExecution();
}
else {
logger.warn(ex,
"Failed to execute IDLE task. " +
"Won't resubmit since not a 'shouldReconnectAutomatically'" +
"or not a 'jakarta.mail.MessagingException'");
ImapIdleChannelAdapter.this.receivingTaskTrigger.stop();
}
publishException(ex);
}
}
@@ -274,21 +280,19 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
Object[] mailMessages = ImapIdleChannelAdapter.this.mailReceiver.receive();
logger.debug(() -> "received " + mailMessages.length + " mail messages");
for (Object mailMessage : mailMessages) {
Runnable messageSendingTask = createMessageSendingTask(mailMessage);
if (isRunning()) {
ImapIdleChannelAdapter.this.sendingTaskExecutor.execute(messageSendingTask);
ImapIdleChannelAdapter.this.messageSender.accept(mailMessage);
}
}
}
}
catch (MessagingException ex) {
catch (jakarta.mail.MessagingException ex) {
logger.warn(ex, "error occurred in idle task");
if (ImapIdleChannelAdapter.this.shouldReconnectAutomatically) {
throw new IllegalStateException("Failure in 'idle' task. Will resubmit.", ex);
}
else {
throw new org.springframework.messaging.MessagingException(
"Failure in 'idle' task. Will NOT resubmit.", ex);
throw new MessagingException("Failure in 'idle' task. Will NOT resubmit.", ex);
}
}
}
@@ -298,7 +302,9 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
private class ExceptionAwarePeriodicTrigger implements Trigger {
private volatile boolean delayNextExecution;
private final AtomicBoolean delayNextExecution = new AtomicBoolean();
private final AtomicBoolean stop = new AtomicBoolean();
ExceptionAwarePeriodicTrigger() {
@@ -306,8 +312,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
@Override
public Instant nextExecution(TriggerContext triggerContext) {
if (this.delayNextExecution) {
this.delayNextExecution = false;
if (this.stop.getAndSet(false)) {
return null;
}
if (this.delayNextExecution.getAndSet(false)) {
return Instant.now().plusMillis(ImapIdleChannelAdapter.this.reconnectDelay);
}
else {
@@ -316,7 +324,11 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
}
void delayNextExecution() {
this.delayNextExecution = true;
this.delayNextExecution.set(true);
}
void stop() {
this.stop.set(true);
}
}

4
spring-integration-mail/src/main/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2023 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.
@@ -54,8 +54,6 @@ public class ImapIdleChannelAdapterParser extends AbstractChannelAdapterParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, txElement,
"synchronization-factory", "transactionSynchronizationFactory");
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor",
"sendingTaskExecutor");
AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(null,
DomUtils.getChildElementByTagName(element, "transactional"), beanDefinition, parserContext);

14
spring-integration-mail/src/main/java/org/springframework/integration/mail/dsl/ImapIdleChannelAdapterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-2023 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.
@@ -22,7 +22,6 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -348,17 +347,6 @@ public class ImapIdleChannelAdapterSpec
return transactional(transactionInterceptor);
}
/**
* Specify a task executor to be used to send messages to the downstream flow.
* @param sendingTaskExecutor the sendingTaskExecutor.
* @return the spec.
* @see ImapIdleChannelAdapter#setSendingTaskExecutor(Executor)
*/
public ImapIdleChannelAdapterSpec sendingTaskExecutor(Executor sendingTaskExecutor) {
this.target.setSendingTaskExecutor(sendingTaskExecutor);
return this;
}
/**
* @param shouldReconnectAutomatically the shouldReconnectAutomatically.
* @return the spec.

20
spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail.xsd
View File

@@ -164,21 +164,6 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to a bean that implements
org.springframework.core.task.TaskExecutor which is used
to send Messages received by this adapter.
If not provided, the adapter uses a single-threaded executor.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.task.TaskExecutor"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cancel-idle-interval" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -197,7 +182,7 @@
<xsd:attribute name="store-uri" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The URI for the Mail Store. Typically of the form: [pop3|imap]://user:password@host:port/INBOX
The URI for the Mail Store. Typically, of the form: [pop3|imap]://user:password@host:port/INBOX
If this is not provided, then the store will be retrieved via the no-arg Session.getStore()
instead of the Session.getStore(url) method.
]]></xsd:documentation>
@@ -206,7 +191,8 @@
<xsd:attribute name="mail-filter-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to provide a SpEL expression which defines a fine grained filtering criteria for the mail messages to be processed by this adapter.
Allows you to provide a SpEL expression which defines a fine-grained
filtering criteria for the mail messages to be processed by this adapter.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>