diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index 57a326ffa0..f33eeb06b5 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -216,6 +216,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource this.fileInfoJson ? file.toJson() : file); } catch (IOException e) { + session.close(); throw new UncheckedIOException("IOException when retrieving " + remotePath, e); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java index 6eb530a89c..df17233b5e 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java @@ -17,22 +17,26 @@ package org.springframework.integration.file.remote; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.io.IOException; import java.io.InputStream; +import java.io.UncheckedIOException; import java.util.Collection; import java.util.Comparator; import java.util.List; import java.util.stream.Collectors; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.file.remote.session.SessionFactory; /** * @author Lukas Gemela @@ -69,6 +73,32 @@ public class RemoteFileStreamingMessageSourceTests { assertThat(testRemoteFileStreamingMessageSource.doReceive()).isNull(); } + @Test + @SuppressWarnings("unchecked") + public void sessionReturnedToCacheProperlyOnDoReceive() throws IOException { + Session session = mock(Session.class); + when(session.readRaw(anyString())).thenThrow(IOException.class); + when(session.list("remoteDirectory")).thenReturn(new String[] { "file1" }); + + SessionFactory sessionFactory = mock(SessionFactory.class); + when(sessionFactory.getSession()).thenReturn(session); + + CachingSessionFactory cachingSessionFactory = new CachingSessionFactory<>(sessionFactory, 1); + RemoteFileTemplate remoteFileTemplate = new RemoteFileTemplate<>(cachingSessionFactory); + + TestRemoteFileStreamingMessageSource testRemoteFileStreamingMessageSource = + new TestRemoteFileStreamingMessageSource(remoteFileTemplate, null); + + testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory"); + testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class)); + testRemoteFileStreamingMessageSource.start(); + + assertThatExceptionOfType(UncheckedIOException.class) + .isThrownBy(testRemoteFileStreamingMessageSource::doReceive); + + assertThat(cachingSessionFactory.getSession()).isNotNull(); + } + static class TestRemoteFileStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource { TestRemoteFileStreamingMessageSource(RemoteFileTemplate template, Comparator comparator) {