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 4db39f36bd..be9d487f17 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 @@ -194,6 +194,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 new file mode 100644 index 0000000000..836dea4f32 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java @@ -0,0 +1,145 @@ +/* + * Copyright 2015-2019 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 + * + * https://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.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.UncheckedIOException; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.BeanFactory; +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 + * @author Artem Bilan + * + * @since 5.2.2 + * + */ +public class RemoteFileStreamingMessageSourceTests { + + @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) { + super(template, comparator); + } + + @Override + protected List> asFileInfoList(Collection files) { + return files + .stream() + .map(TestFileInfo::new) + .collect(Collectors.toList()); + } + + @Override + protected boolean isDirectory(String file) { + return false; + } + + @Override + public String getComponentType() { + return null; + } + + } + + static class TestFileInfo extends AbstractFileInfo { + + TestFileInfo(String fileName) { + this.fileName = fileName; + } + + private final String fileName; + + @Override + public boolean isDirectory() { + return false; + } + + @Override + public boolean isLink() { + return false; + } + + @Override + public long getSize() { + return 0; + } + + @Override + public long getModified() { + return 0; + } + + @Override + public String getFilename() { + return fileName; + } + + @Override + public String getPermissions() { + return null; + } + + @Override + public String getFileInfo() { + return null; + } + + } + +}