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

# Conflicts:
#	spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
#	spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
#	spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
#	spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
#	spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
#	spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
This commit is contained in:
Gary Russell
2019-03-01 14:35:31 -05:00
committed by Artem Bilan
parent 442b3d9471
commit 45789a99b8
3 changed files with 115 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 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.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -26,15 +27,18 @@ import java.util.Iterator;
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.AbstractMessageSource;
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.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -47,8 +51,8 @@ import org.springframework.util.Assert;
* @since 4.3
*
*/
public abstract class AbstractRemoteFileStreamingMessageSource<F> extends AbstractMessageSource<InputStream>
implements BeanFactoryAware, InitializingBean {
public abstract class AbstractRemoteFileStreamingMessageSource<F>
extends AbstractMessageSource<InputStream> implements Lifecycle {
private final RemoteFileTemplate<F> remoteFileTemplate;
@@ -56,6 +60,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
private final Comparator<AbstractFileInfo<F>> comparator;
private final AtomicBoolean running = new AtomicBoolean();
/**
* the path on the remote server.
*/
@@ -107,7 +113,11 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
* @param filter the file list filter.
*/
public void setFilter(FileListFilter<F> filter) {
this.filter = filter;
doSetFilter(filter);
}
protected final void doSetFilter(FileListFilter<F> filterToSet) {
this.filter = filterToSet;
}
protected RemoteFileTemplate<F> getRemoteFileTemplate() {
@@ -127,26 +137,67 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F> extends Abstra
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())
.build();
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())
.build();
}
catch (IOException e) {
throw new MessagingException("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;
}
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();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 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.
@@ -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.Rule;
import org.junit.Test;
@@ -41,15 +44,22 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
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
* @author Artem Bilan
*
* @since 4.3
*
*/
@@ -67,6 +77,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());
assertEquals("foo\nbar", new String(received.getPayload()));
assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY));
@@ -90,16 +101,17 @@ public class StreamingInboundTests {
streamer.setBeanFactory(mock(BeanFactory.class));
streamer.setRemoteDirectory("/bad");
streamer.afterPropertiesSet();
streamer.receive();
streamer.start();
}
@SuppressWarnings("unchecked")
@Test
@SuppressWarnings("unchecked")
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));
@@ -140,8 +152,11 @@ public class StreamingInboundTests {
public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {
ConcurrentHashMap<String, String> metadataMap = new ConcurrentHashMap<>();
protected Streamer(RemoteFileTemplate<String> template, Comparator<AbstractFileInfo<String>> comparator) {
super(template, comparator);
doSetFilter(new StringPersistentFileListFilter(new SimpleMetadataStore(this.metadataMap), "streamer"));
}
@Override
@@ -151,7 +166,7 @@ public class StreamingInboundTests {
@Override
protected List<AbstractFileInfo<String>> asFileInfoList(Collection<String> files) {
List<AbstractFileInfo<String>> infos = new ArrayList<AbstractFileInfo<String>>();
List<AbstractFileInfo<String>> infos = new ArrayList<>();
for (String file : files) {
infos.add(new StringFileInfo(file));
}
@@ -200,7 +215,7 @@ public class StreamingInboundTests {
@Override
public String getFileInfo() {
return null;
return name;
}
}
@@ -215,9 +230,14 @@ public class StreamingInboundTests {
public static class StringSessionFactory implements SessionFactory<String> {
private Session<String> singletonSession;
@SuppressWarnings("unchecked")
@Override
public Session<String> getSession() {
if (this.singletonSession != null) {
return this.singletonSession;
}
try {
Session<String> session = mock(Session.class);
willReturn(new String[] { "/foo/foo", "/foo/bar" }).given(session).list("/foo");
@@ -232,10 +252,14 @@ 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.singletonSession = session;
return session;
}
catch (Exception e) {
@@ -245,4 +269,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;
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 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.