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

GH-2776: Fix Streaming Remote File MessageSource

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/2776
Also see https://github.com/spring-projects/spring-integration/issues/2777

- reset the filter for the current file if the fetch fails
- implement `Lifecycle` and clear the `toBeReceived` queue and corresponding
  filter entries

* Polishing - PR Comments **cherry-pick to all supported**

* Polishing
This commit is contained in:
Gary Russell
2019-03-01 14:35:31 -05:00
committed by Artem Bilan
parent 23843c789b
commit a167290f9b
5 changed files with 150 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-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.
@@ -18,6 +18,7 @@ package org.springframework.integration.file.remote;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -25,20 +26,21 @@ import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.filters.ResettableFileListFilter;
import org.springframework.integration.file.filters.ReversibleFileListFilter;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.support.FileUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -53,7 +55,7 @@ import org.springframework.util.ObjectUtils;
*
*/
public abstract class AbstractRemoteFileStreamingMessageSource<F>
extends AbstractFetchLimitingMessageSource<InputStream> implements BeanFactoryAware, InitializingBean {
extends AbstractFetchLimitingMessageSource<InputStream> implements Lifecycle {
private final RemoteFileTemplate<F> remoteFileTemplate;
@@ -61,6 +63,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
private final Comparator<F> comparator;
private final AtomicBoolean running = new AtomicBoolean();
private boolean fileInfoJson = true;
/**
@@ -117,8 +121,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
doSetFilter(filter);
}
protected final void doSetFilter(FileListFilter<F> filter) {
this.filter = filter;
protected final void doSetFilter(FileListFilter<F> filterToSet) {
this.filter = filterToSet;
}
/**
@@ -149,23 +153,53 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
protected void doInit() {
}
@Override
public void start() {
this.running.set(true);
}
@Override
public void stop() {
if (this.running.compareAndSet(true, false)) {
// remove unprocessed files from the queue (and filter)
AbstractFileInfo<F> file = this.toBeReceived.poll();
while (file != null) {
resetFilterIfNecessary(file);
file = this.toBeReceived.poll();
}
}
}
@Override
public boolean isRunning() {
return this.running.get();
}
@Override
protected Object doReceive() {
Assert.state(this.running.get(), () -> getComponentName() + " is not running");
AbstractFileInfo<F> file = poll();
if (file != null) {
String remotePath = remotePath(file);
Session<?> session = this.remoteFileTemplate.getSession();
try {
return getMessageBuilderFactory()
.withPayload(session.readRaw(remotePath))
.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
.setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
.setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
.setHeader(FileHeaders.REMOTE_FILE_INFO,
this.fileInfoJson ? file.toJson() : file);
String remotePath = remotePath(file);
Session<?> session = this.remoteFileTemplate.getSession();
try {
return getMessageBuilderFactory()
.withPayload(session.readRaw(remotePath))
.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
.setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
.setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
.setHeader(FileHeaders.REMOTE_FILE_INFO,
this.fileInfoJson ? file.toJson() : file);
}
catch (IOException e) {
throw new UncheckedIOException("IOException when retrieving " + remotePath, e);
}
}
catch (IOException e) {
throw new MessagingException("IOException when retrieving " + remotePath, e);
catch (RuntimeException e) {
resetFilterIfNecessary(file);
throw e;
}
}
return null;
@@ -176,6 +210,16 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
return doReceive();
}
private void resetFilterIfNecessary(AbstractFileInfo<F> file) {
if (this.filter instanceof ResettableFileListFilter) {
if (this.logger.isInfoEnabled()) {
this.logger.info("Removing the remote file '" + file +
"' from the filter for a subsequent transfer attempt");
}
((ResettableFileListFilter<F>) this.filter).remove(file.getFileInfo());
}
}
protected AbstractFileInfo<F> poll() {
if (this.toBeReceived.size() == 0) {
listFiles();

87
spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
View File

@@ -28,10 +28,13 @@ import static org.mockito.Mockito.verify;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import org.junit.Test;
@@ -40,13 +43,16 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.splitter.FileSplitter;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.StreamTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
/**
* @author Gary Russell
@@ -67,6 +73,7 @@ public class StreamingInboundTests {
streamer.setBeanFactory(mock(BeanFactory.class));
streamer.setRemoteDirectory("/foo");
streamer.afterPropertiesSet();
streamer.start();
Message<byte[]> received = (Message<byte[]>) this.transformer.transform(streamer.receive());
assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes());
assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo");
@@ -112,6 +119,7 @@ public class StreamingInboundTests {
streamer.setMaxFetchSize(1);
streamer.setFilter(new AcceptOnceFileListFilter<>());
streamer.afterPropertiesSet();
streamer.start();
Message<byte[]> received = (Message<byte[]>) this.transformer.transform(streamer.receive());
assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes());
assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo");
@@ -138,17 +146,18 @@ public class StreamingInboundTests {
streamer.setBeanFactory(mock(BeanFactory.class));
streamer.setRemoteDirectory("/bad");
streamer.afterPropertiesSet();
assertThatExceptionOfType(MessagingException.class)
streamer.start();
assertThatExceptionOfType(UncheckedIOException.class)
.isThrownBy(streamer::receive);
}
@SuppressWarnings("unchecked")
@Test
public void testLineByLine() throws Exception {
Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null);
streamer.setBeanFactory(mock(BeanFactory.class));
streamer.setRemoteDirectory("/foo");
streamer.afterPropertiesSet();
streamer.start();
QueueChannel out = new QueueChannel();
FileSplitter splitter = new FileSplitter();
splitter.setBeanFactory(mock(BeanFactory.class));
@@ -160,7 +169,7 @@ public class StreamingInboundTests {
assertThat(received.getPayload()).isEqualTo("foo");
assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo");
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo");
received = (Message<byte[]>) out.receive(0);
received = out.receive(0);
assertThat(received.getPayload()).isEqualTo("bar");
assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo");
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo");
@@ -185,10 +194,43 @@ public class StreamingInboundTests {
verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(5)).close();
}
@SuppressWarnings("unchecked")
@Test
public void testStopAdapterRemovesUnprocessed() {
Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null);
streamer.setBeanFactory(mock(BeanFactory.class));
streamer.setRemoteDirectory("/foo");
streamer.afterPropertiesSet();
streamer.start();
assertThat(streamer.receive()).isNotNull();
assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1);
assertThat(streamer.metadataMap).hasSize(2);
streamer.stop();
assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(0);
assertThat(streamer.metadataMap).hasSize(1);
}
@SuppressWarnings("unchecked")
@Test
public void testFilterReversedOnBadFetch() {
Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null);
streamer.setBeanFactory(mock(BeanFactory.class));
streamer.setRemoteDirectory("/bad");
streamer.afterPropertiesSet();
streamer.start();
assertThatExceptionOfType(UncheckedIOException.class)
.isThrownBy(streamer::receive);
assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1);
assertThat(streamer.metadataMap).hasSize(1);
}
public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {
ConcurrentHashMap<String, String> metadataMap = new ConcurrentHashMap<>();
protected Streamer(RemoteFileTemplate<String> template, Comparator<String> comparator) {
super(template, comparator);
doSetFilter(new StringPersistentFileListFilter(new SimpleMetadataStore(this.metadataMap), "streamer"));
}
@Override
@@ -252,11 +294,7 @@ public class StreamingInboundTests {
@Override
public String getFileInfo() {
return asString();
}
private String asString() {
return "StringFileInfo [name=" + this.name + "]";
return name;
}
}
@@ -271,13 +309,13 @@ public class StreamingInboundTests {
public static class StringSessionFactory implements SessionFactory<String> {
private Session<String> session;
private Session<String> singletonSession;
@SuppressWarnings("unchecked")
@Override
public Session<String> getSession() {
if (this.session != null) {
return this.session;
if (this.singletonSession != null) {
return this.singletonSession;
}
try {
Session<String> session = mock(Session.class);
@@ -293,12 +331,13 @@ public class StreamingInboundTests {
willReturn(foo2).given(session).readRaw("/bar/foo");
willReturn(bar2).given(session).readRaw("/bar/bar");
willReturn(new String[] { "/bad/file" }).given(session).list("/bad");
willThrow(new IOException("No file")).given(session).readRaw("/bad/file");
willReturn(new String[] { "/bad/file1", "/bad/file2" }).given(session).list("/bad");
willThrow(new IOException("No file")).given(session).readRaw("/bad/file1");
willThrow(new IOException("No file")).given(session).readRaw("/bad/file2");
given(session.finalizeRaw()).willReturn(true);
this.session = session;
this.singletonSession = session;
return session;
}
@@ -309,4 +348,22 @@ public class StreamingInboundTests {
}
public static class StringPersistentFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<String> {
public StringPersistentFileListFilter(ConcurrentMetadataStore store, String prefix) {
super(store, prefix);
}
@Override
protected long modified(String file) {
return 0;
}
@Override
protected String fileName(String file) {
return file;
}
}
}

14
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
View File

@@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.io.Closeable;
import java.io.InputStream;
import java.util.Comparator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -48,6 +49,8 @@ import org.springframework.integration.ftp.session.FtpFileInfo;
import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.util.OnlyOnceTrigger;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.StreamTransformer;
import org.springframework.messaging.Message;
import org.springframework.scheduling.support.PeriodicTrigger;
@@ -114,12 +117,17 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
this.source.setFileInfoJson(false);
this.data.purge(null);
this.metadataMap.clear();
this.adapter.setTrigger(new OnlyOnceTrigger());
this.adapter.setMaxMessagesPerPoll(1);
this.adapter.start();
assertThat(this.data.receive(10000)).isNotNull();
received = (Message<byte[]>) this.data.receive(10000);
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO)).isInstanceOf(FtpFileInfo.class);
assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).hasSize(1);
assertThat(this.metadataMap).hasSize(2);
this.adapter.stop();
assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).isEmpty();
assertThat(this.metadataMap).hasSize(1);
}
@Test
@@ -127,6 +135,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
FtpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.afterPropertiesSet();
messageSource.start();
Message<InputStream> received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo(" ftpSource1.txt");
@@ -142,6 +151,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
FtpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(null);
messageSource.afterPropertiesSet();
messageSource.start();
Message<InputStream> received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo(" ftpSource1.txt");
@@ -174,7 +184,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
public PollerMetadata defaultPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(500));
pollerMetadata.setMaxMessagesPerPoll(1);
pollerMetadata.setMaxMessagesPerPoll(2);
return pollerMetadata;
}

3
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java
View File

@@ -77,7 +77,8 @@ public class StoredProcJavaConfigTests {
public void test() {
Message<?> received = fooChannel.receive(10000);
assertThat(received).isNotNull();
Collection<?> primes = (Collection<?>) received.getPayload();
@SuppressWarnings("unchecked")
Collection<Integer> primes = (Collection<Integer>) received.getPayload();
assertThat(primes).containsExactly(2, 3, 5, 7);
received = fooChannel.receive(100);
// verify maxMessagesPerPoll == 1

3
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
View File

@@ -127,6 +127,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
SftpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.afterPropertiesSet();
messageSource.start();
Message<InputStream> received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE))
@@ -140,6 +141,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
SftpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(null);
messageSource.afterPropertiesSet();
messageSource.start();
Message<InputStream> received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE))
@@ -153,6 +155,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
SftpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(Arrays::asList);
messageSource.afterPropertiesSet();
messageSource.start();
Message<InputStream> received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE))