INT-4299: Add AbstractMailReceiver.autoCloseFolder (#2887)

* INT-4299: Add AbstractMailReceiver.autoCloseFolder

JIRA: https://jira.spring.io/browse/INT-4299

* Optimize `AbstractMessageSource` do not evaluate headers when polling
result is `null`
* Optimize `MailReceivingMessageSource` do not wrap polling result to
the message: the `AbstractMessageSource` will do that later

* * Fix JavaDocs
* Document `autoCloseFolder` option

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java

@@ -162,6 +162,9 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 
 	@SuppressWarnings("unchecked")
 	protected Message<T> buildMessage(Object result) {
+		if (result == null) {
+			return null;
+		}
 		Message<?> message = null;
 		Map<String, Object> headers = evaluateHeaders();
 		if (result instanceof AbstractIntegrationMessageBuilder<?>) {
@@ -181,14 +184,14 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 								.build();
 			}
 		}
-		else if (result != null) {
+		else {
 			message =
 					getMessageBuilderFactory()
 							.withPayload(result)
 							.copyHeaders(headers)
 							.build();
 		}
-		if (this.countsEnabled && message != null) {
+		if (this.countsEnabled) {
 			if (this.metricsCaptor != null) {
 				incrementReceiveCounter();
 			}

spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.mail;
 
 import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Date;
@@ -43,9 +44,11 @@ import javax.mail.internet.MimeMessage;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.mapping.HeaderMapper;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.FileCopyUtils;
@@ -103,6 +106,8 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 
 	private boolean simpleContent;
 
+	private boolean autoCloseFolder = true;
+
 	private volatile Store store;
 
 	private volatile Folder folder;
@@ -266,6 +271,22 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		this.simpleContent = simpleContent;
 	}
 
+	/**
+	 * Configure a {@code boolean} flag to close the folder automatically after a fetch (default) or
+	 * populate an additional {@link IntegrationMessageHeaderAccessor#CLOSEABLE_RESOURCE} message header instead.
+	 * It is the downstream flow's responsibility to obtain this header and call its {@code close()} whenever
+	 * it is necessary.
+	 * <p> Keeping the folder open is useful in cases where communication with the server is needed
+	 * when parsing multipart content of the email with attachments.
+	 * <p> The {@link #setSimpleContent(boolean)} and {@link #setHeaderMapper(HeaderMapper)} options are not
+	 * affected by this flag.
+	 * @param autoCloseFolder {@code false} do not close the folder automatically after a fetch.
+	 * @since 5.2
+	 */
+	public void setAutoCloseFolder(boolean autoCloseFolder) {
+		this.autoCloseFolder = autoCloseFolder;
+	}
+
 	protected Folder getFolder() {
 		return this.folder;
 	}
@@ -357,7 +378,9 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 				return convertMessagesIfNecessary(searchAndFilterMessages());
 			}
 			finally {
-				MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
+				if (this.autoCloseFolder) {
+					closeFolder();
+				}
 			}
 		}
 		finally {
@@ -367,6 +390,16 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 		}
 	}
 
+	private void closeFolder() {
+		this.folderReadLock.lock();
+		try {
+			MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
+		}
+		finally {
+			this.folderReadLock.unlock();
+		}
+	}
+
 	private MimeMessage[] searchAndFilterMessages() throws MessagingException {
 		if (this.logger.isInfoEnabled()) {
 			this.logger.info("attempting to receive mail from folder [" + this.folder.getFullName() + "]");
@@ -395,17 +428,26 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	}
 
 	private Object[] convertMessagesIfNecessary(MimeMessage[] filteredMessages) {
-		if (this.headerMapper != null) {
+		if (this.headerMapper != null || !this.autoCloseFolder) {
 			org.springframework.messaging.Message<?>[] converted =
 					new org.springframework.messaging.Message<?>[filteredMessages.length];
 			int n = 0;
 			for (MimeMessage message : filteredMessages) {
-				Map<String, Object> headers = this.headerMapper.toHeaders(message);
-				converted[n++] =
+				Object payload = message;
+				Map<String, Object> headers = null;
+				if (this.headerMapper != null) {
+					headers = this.headerMapper.toHeaders(message);
+					payload = extractContent(message, headers);
+				}
+				AbstractIntegrationMessageBuilder<Object> messageBuilder =
 						getMessageBuilderFactory()
-								.withPayload(extractContent(message, headers))
-								.copyHeaders(headers)
-								.build();
+								.withPayload(payload)
+								.copyHeaders(headers);
+				if (!this.autoCloseFolder) {
+					messageBuilder.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE,
+							(Closeable) this::closeFolder);
+				}
+				converted[n++] = messageBuilder.build();
 			}
 			return converted;
 		}
@@ -566,7 +608,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 	public void destroy() {
 		this.folderWriteLock.lock();
 		try {
-			MailTransportUtils.closeFolder(this.folder, this.shouldDeleteMessages);
+			closeFolder();
 			MailTransportUtils.closeService(this.store);
 			this.folder = null;
 			this.store = null;
@@ -628,11 +670,16 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
 
 		@Override
 		public Folder getFolder() {
-			try {
-				return obtainFolderInstance();
+			if (!AbstractMailReceiver.this.autoCloseFolder) {
+				return AbstractMailReceiver.this.folder;
 			}
-			catch (MessagingException e) {
-				throw new org.springframework.messaging.MessagingException("Unable to obtain the mail folder", e);
+			else {
+				try {
+					return obtainFolderInstance();
+				}
+				catch (MessagingException e) {
+					throw new org.springframework.messaging.MessagingException("Unable to obtain the mail folder", e);
+				}
 			}
 		}
 

spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java

@@ -21,14 +21,12 @@ import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.springframework.integration.endpoint.AbstractMessageSource;
-import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
 /**
  * {@link org.springframework.integration.core.MessageSource} implementation that
- * delegates to a
- * {@link MailReceiver} to poll a mailbox. Each poll of the mailbox may
+ * delegates to a {@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
@@ -69,12 +67,7 @@ public class MailReceivingMessageSource extends AbstractMessageSource<Object> {
 				if (this.logger.isDebugEnabled()) {
 					this.logger.debug("received mail message [" + mailMessage + "]");
 				}
-				if (mailMessage instanceof Message) {
-					return getMessageBuilderFactory().fromMessage((Message<?>) mailMessage);
-				}
-				else {
-					return getMessageBuilderFactory().withPayload(mailMessage);
-				}
+				return mailMessage;
 			}
 		}
 		catch (Exception e) {

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

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.mail.ImapIdleChannelAdapter;
@@ -37,6 +38,7 @@ import org.springframework.util.xml.DomUtils;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class ImapIdleChannelAdapterParser extends AbstractChannelAdapterParser {
 
@@ -52,7 +54,8 @@ public class ImapIdleChannelAdapterParser extends AbstractChannelAdapterParser {
 			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, txElement,
 					"synchronization-factory", "transactionSynchronizationFactory");
 		}
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor", "sendingTaskExecutor");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor",
+				"sendingTaskExecutor");
 		AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
 		IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(null,
 				DomUtils.getChildElementByTagName(element, "transactional"), beanDefinition, parserContext);
@@ -90,15 +93,15 @@ public class ImapIdleChannelAdapterParser extends AbstractChannelAdapterParser {
 
 		String selectorExpression = element.getAttribute("mail-filter-expression");
 
-		RootBeanDefinition expressionDef = null;
 		if (StringUtils.hasText(selectorExpression)) {
-			expressionDef = new RootBeanDefinition("org.springframework.integration.config.ExpressionFactoryBean");
+			RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
 			expressionDef.getConstructorArgumentValues().addGenericArgumentValue(selectorExpression);
 			receiverBuilder.addPropertyValue("selectorExpression", expressionDef);
 		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(receiverBuilder, element, "header-mapper");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "embedded-parts-as-bytes");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "simple-content");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "auto-close-folder");
 
 		return receiverBuilder.getBeanDefinition();
 	}

spring-integration-mail/src/main/java/org/springframework/integration/mail/config/MailInboundChannelAdapterParser.java

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.mail.MailReceivingMessageSource;
@@ -36,6 +37,7 @@ import org.springframework.util.xml.DomUtils;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class MailInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
 
@@ -87,15 +89,15 @@ public class MailInboundChannelAdapterParser extends AbstractPollingInboundChann
 
 		String selectorExpression = element.getAttribute("mail-filter-expression");
 
-		RootBeanDefinition expressionDef = null;
 		if (StringUtils.hasText(selectorExpression)) {
-			expressionDef = new RootBeanDefinition("org.springframework.integration.config.ExpressionFactoryBean");
+			RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
 			expressionDef.getConstructorArgumentValues().addGenericArgumentValue(selectorExpression);
 			receiverBuilder.addPropertyValue("selectorExpression", expressionDef);
 		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(receiverBuilder, element, "header-mapper");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "embedded-parts-as-bytes");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "simple-content");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(receiverBuilder, element, "auto-close-folder");
 
 		return receiverBuilder.getBeanDefinition();
 	}

spring-integration-mail/src/main/java/org/springframework/integration/mail/config/MailReceiverFactoryBean.java

@@ -81,6 +81,8 @@ public class MailReceiverFactoryBean extends AbstractFactoryBean<MailReceiver> {
 
 	private Boolean simpleContent;
 
+	private Boolean autoCloseFolder;
+
 	public void setStoreUri(@Nullable String storeUri) {
 		this.storeUri = storeUri;
 	}
@@ -141,6 +143,10 @@ public class MailReceiverFactoryBean extends AbstractFactoryBean<MailReceiver> {
 		this.simpleContent = simpleContent;
 	}
 
+	public void setAutoCloseFolder(@Nullable Boolean autoCloseFolder) {
+		this.autoCloseFolder = autoCloseFolder;
+	}
+
 	@Override
 	protected MailReceiver createInstance() {
 		if (this.receiver == null) {
@@ -211,6 +217,9 @@ public class MailReceiverFactoryBean extends AbstractFactoryBean<MailReceiver> {
 		if (this.simpleContent != null) {
 			mailReceiver.setSimpleContent(this.simpleContent);
 		}
+		if (this.autoCloseFolder != null) {
+			mailReceiver.setAutoCloseFolder(this.autoCloseFolder);
+		}
 		mailReceiver.afterPropertiesSet();
 		return mailReceiver;
 	}

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

@@ -269,6 +269,21 @@ public class ImapIdleChannelAdapterSpec
 		return _this();
 	}
 
+	/**
+	 * When configured to {@code false}, the folder is not closed automatically after a fetch.
+	 * It is the target application's responsibility to close it using the
+	 * {@link org.springframework.integration.IntegrationMessageHeaderAccessor#CLOSEABLE_RESOURCE} header
+	 * from the message produced by this channel adapter.
+	 * @param autoCloseFolder set to {@code false} to keep folder opened.
+	 * @return the spec.
+	 * @since 5.2
+	 * @see ImapMailReceiver#setAutoCloseFolder(boolean)
+	 */
+	public ImapIdleChannelAdapterSpec autoCloseFolder(boolean autoCloseFolder) {
+		assertReceiver();
+		this.receiver.setAutoCloseFolder(autoCloseFolder);
+		return _this();
+	}
 
 	/**
 	 * Configure a {@link TransactionSynchronizationFactory}. Usually used to synchronize

spring-integration-mail/src/main/java/org/springframework/integration/mail/dsl/MailInboundChannelAdapterSpec.java

@@ -253,6 +253,22 @@ MailInboundChannelAdapterSpec<S extends MailInboundChannelAdapterSpec<S, R>, R e
 		return _this();
 	}
 
+	/**
+	 * When configured to {@code false}, the folder is not closed automatically after a fetch.
+	 * It is the target application's responsibility to close it using the
+	 * {@link org.springframework.integration.IntegrationMessageHeaderAccessor#CLOSEABLE_RESOURCE} header
+	 * from the message produced by this channel adapter.
+	 * @param autoCloseFolder set to {@code false} to keep folder opened.
+	 * @return the spec.
+	 * @since 5.2
+	 * @see AbstractMailReceiver#setAutoCloseFolder(boolean)
+	 */
+	public S autoCloseFolder(boolean autoCloseFolder) {
+		assertReceiver();
+		this.receiver.setAutoCloseFolder(autoCloseFolder);
+		return _this();
+	}
+
 	@Override
 	public Map<Object, String> getComponentsToRegister() {
 		return Collections.singletonMap(this.receiver, this.receiver.getComponentName());

spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail-5.2.xsd

@@ -416,6 +416,16 @@
 				<xsd:union memberTypes="xsd:boolean xsd:string"/>
 			</xsd:simpleType>
 		</xsd:attribute>
+		<xsd:attribute name="auto-close-folder" default="true">
+			<xsd:annotation>
+				<xsd:documentation>
+					Default 'true' - close folder automatically after fetch, before producing messages.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 	</xsd:attributeGroup>
 
 </xsd:schema>

spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests-context.xml

@@ -38,6 +38,7 @@
 			header-mapper="mapper"
 			simple-content="true"
 			embedded-parts-as-bytes="false"
+			auto-close-folder="false"
 			should-delete-messages="true"/>
 
 	<bean id="mapper" class="org.springframework.integration.mail.support.DefaultMailHeaderMapper" />

spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests.java

@@ -84,6 +84,7 @@ public class ImapIdleChannelAdapterParserTests {
 		assertThat(receiverAccessor.getPropertyValue("embeddedPartsAsBytes")).isEqualTo(Boolean.FALSE);
 		assertThat(receiverAccessor.getPropertyValue("headerMapper")).isNotNull();
 		assertThat(receiverAccessor.getPropertyValue("simpleContent")).isEqualTo(Boolean.TRUE);
+		assertThat(receiverAccessor.getPropertyValue("autoCloseFolder")).isEqualTo(false);
 	}
 
 	@Test

spring-integration-mail/src/test/java/org/springframework/integration/mail/config/InboundChannelAdapterParserTests-context.xml

@@ -16,6 +16,7 @@
 	<mail:inbound-channel-adapter id="pop3ShouldDeleteTrue"
 		store-uri="pop3:test" channel="testChannel" should-delete-messages="true"
 		header-mapper="mapper" embedded-parts-as-bytes="false"
+		auto-close-folder="false"
 		auto-startup="false" />
 
 	<mail:inbound-channel-adapter id="pop3ShouldDeleteFalse"

spring-integration-mail/src/test/java/org/springframework/integration/mail/config/InboundChannelAdapterParserTests.java

@@ -74,6 +74,7 @@ public class InboundChannelAdapterParserTests {
 		Boolean value = (Boolean) receiverAccessor.getPropertyValue("shouldDeleteMessages");
 		assertThat(value).isTrue();
 		assertThat(receiverAccessor.getPropertyValue("embeddedPartsAsBytes")).isEqualTo(Boolean.FALSE);
+		assertThat(receiverAccessor.getPropertyValue("autoCloseFolder")).isEqualTo(false);
 		assertThat(receiverAccessor.getPropertyValue("headerMapper")).isNotNull();
 	}
 

spring-integration-mail/src/test/java/org/springframework/integration/mail/dsl/MailTests.java

@@ -18,6 +18,8 @@ package org.springframework.integration.mail.dsl;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.Properties;
 
 import javax.mail.Flags;
@@ -40,6 +42,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
@@ -60,15 +63,13 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Gary Russell
  * @author Artem Bilan
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class MailTests {
 
@@ -144,14 +145,16 @@ public class MailTests {
 	}
 
 	@Test
-	public void testPop3() throws Exception {
+	public void testPop3() throws IOException {
 		Message<?> message = this.pop3Channel.receive(10000);
 		assertThat(message).isNotNull();
 		MessageHeaders headers = message.getHeaders();
-		assertThat(headers.get(MailHeaders.TO, String[].class)[0]).isEqualTo("Foo <foo@bar>");
+		assertThat(headers.get(MailHeaders.TO, String[].class)).containsExactly("Foo <foo@bar>");
 		assertThat(headers.get(MailHeaders.FROM)).isEqualTo("Bar <bar@baz>");
 		assertThat(headers.get(MailHeaders.SUBJECT)).isEqualTo("Test Email");
 		assertThat(message.getPayload()).isEqualTo("foo\r\n\r\n");
+		assertThat(message.getHeaders().containsKey(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE)).isTrue();
+		message.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, Closeable.class).close();
 	}
 
 	@Test
@@ -163,6 +166,8 @@ public class MailTests {
 		assertThat(mm.getFrom()[0].toString()).isEqualTo("Bar <bar@baz>");
 		assertThat(mm.getSubject()).isEqualTo("Test Email");
 		assertThat(mm.getContent()).isEqualTo(TestMailServer.MailServer.MailHandler.BODY + "\r\n");
+		assertThat(message.getHeaders().containsKey(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE)).isTrue();
+		message.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, Closeable.class).close();
 	}
 
 	@Test
@@ -170,10 +175,11 @@ public class MailTests {
 		Message<?> message = this.imapIdleChannel.receive(10000);
 		assertThat(message).isNotNull();
 		MessageHeaders headers = message.getHeaders();
-		assertThat(headers.get(MailHeaders.TO, String[].class)[0]).isEqualTo("Foo <foo@bar>");
+		assertThat(headers.get(MailHeaders.TO, String[].class)).containsExactly("Foo <foo@bar>");
 		assertThat(headers.get(MailHeaders.FROM)).isEqualTo("Bar <bar@baz>");
 		assertThat(headers.get(MailHeaders.SUBJECT)).isEqualTo("Test Email");
 		assertThat(message.getPayload()).isEqualTo(TestMailServer.MailServer.MailHandler.MESSAGE + "\r\n");
+		assertThat(message.getHeaders().containsKey(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE)).isTrue();
 		this.imapIdleAdapter.stop();
 		assertThat(TestUtils.getPropertyValue(this.imapIdleAdapter, "shouldReconnectAutomatically", Boolean.class))
 				.isFalse();
@@ -204,6 +210,7 @@ public class MailTests {
 			return IntegrationFlows
 					.from(Mail.pop3InboundAdapter("localhost", pop3Server.getPort(), "user", "pw")
 									.javaMailProperties(p -> p.put("mail.debug", "false"))
+									.autoCloseFolder(false)
 									.headerMapper(mailHeaderMapper()),
 							e -> e.autoStartup(true).poller(p -> p.fixedDelay(1000)))
 					.enrichHeaders(s -> s.headerExpressions(c -> c.put(MailHeaders.SUBJECT, "payload.subject")
@@ -218,6 +225,7 @@ public class MailTests {
 					.from(Mail.imapInboundAdapter("imap://user:pw@localhost:" + imapServer.getPort() + "/INBOX")
 									.searchTermStrategy(this::fromAndNotSeenTerm)
 									.userFlag("testSIUserFlag")
+									.autoCloseFolder(false)
 									.simpleContent(true)
 									.javaMailProperties(p -> p.put("mail.debug", "false")),
 							e -> e.autoStartup(true)
@@ -230,13 +238,14 @@ public class MailTests {
 		public IntegrationFlow imapIdleFlow() {
 			return IntegrationFlows
 					.from(Mail.imapIdleAdapter("imap://user:pw@localhost:" + imapIdleServer.getPort() + "/INBOX")
-									.autoStartup(true)
-									.searchTermStrategy(this::fromAndNotSeenTerm)
-									.userFlag("testSIUserFlag")
-									.javaMailProperties(p -> p.put("mail.debug", "false")
-											.put("mail.imap.connectionpoolsize", "5"))
-									.shouldReconnectAutomatically(false)
-									.headerMapper(mailHeaderMapper()))
+							.autoStartup(true)
+							.searchTermStrategy(this::fromAndNotSeenTerm)
+							.userFlag("testSIUserFlag")
+							.autoCloseFolder(false)
+							.javaMailProperties(p -> p.put("mail.debug", "false")
+									.put("mail.imap.connectionpoolsize", "5"))
+							.shouldReconnectAutomatically(false)
+							.headerMapper(mailHeaderMapper()))
 					.channel(MessageChannels.queue("imapIdleChannel"))
 					.get();
 		}

src/reference/asciidoc/ftp.adoc

@@ -596,7 +596,7 @@ consumed.
 The session is provided in the `closeableResource` header (`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE`).
 Standard framework components, such as the `FileSplitter` and `StreamTransformer`, automatically close the session.
 See <<file-splitter>> and <<stream-transformer>> for more information about these components.
-The follwoing example shows how to configure an `inbound-streaming-channel-adapter`:
+The following example shows how to configure an `inbound-streaming-channel-adapter`:
 
 ====
 [source, xml]

src/reference/asciidoc/mail.adoc

@@ -111,11 +111,30 @@ Starting with version 2.2, the framework eagerly fetches IMAP messages and expos
 This had the undesired side effect of changing the `getContent()` behavior.
 This inconsistency was further exacerbated by the <<mail-mapping, Mail Mapping>> enhancement introduced in version 4.3, because, when a header mapper was provided, the payload was rendered by the `IMAPMessage.getContent()` method.
 This meant that the IMAP content differed, depending on whether or not a header mapper was provided.
+
 Starting with version 5.0, messages originating from an IMAP source render the content in accordance with `IMAPMessage.getContent()` behavior, regardless of whether a header mapper is provided.
 If you do not use a header mapper and you wish to revert to the previous behavior of rendering only the body, set the `simpleContent` boolean property on the mail receiver to `true`.
 This property now controls the rendering regardless of whether a header mapper is used.
 It now allows body-only rendering when a header mapper is provided.
 
+Starting with version 5.2, the `autoCloseFolder` option is provided on the mail receiver.
+Setting it to `false` doesn't close the folder automatically after a fetch, but instead an `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header (see <<message-header-accessor>> for more information) is populated into every message to producer from the channel adapter.
+It is the target application's responsibility to call the `close()` on this header whenever it is necessary in the downstream flow:
+
+====
+[source,java]
+----
+Closeable closeableResource = StaticMessageHeaderAccessor.getCloseableResource(mailMessage);
+if (closeableResource != null) {
+    closeableResource.close();
+}
+
+----
+====
+
+Keeping the folder open is useful in cases where communication with the server is needed during parsing multipart content of the email with attachments.
+The `close()` on the `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header delegates to the `AbstractMailReceiver` to close the folder with `expunge` option if `shouldDeleteMessages` is configured respectively on the `AbstractMailReceiver`.
+
 [[mail-mapping]]
 === Inbound Mail Message Mapping
 

src/reference/asciidoc/whats-new.adoc

@@ -42,3 +42,9 @@ See <<tcp-codecs>> for more information.
 
 When using a `TcpNioServerConnectionFactory`, priority is now given to accepting new connections over reading from existing connections, but it is configurable.
 See <<note-nio>> for more information.
+
+[[x5.2-mail]]
+==== Mail Changes
+
+The `AbstractMailReceiver` has now an `autoCloseFolder` option (`true` by default), to disable an automatic folder close after a fetch, but populate `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header instead for downstream interaction.
+See <<mail-inbound>> for more information.