FileInbound DSL: Add recursive for convenience (#3495)
* FileInbound DSL: Add `recursive` for convenience Related to https://stackoverflow.com/questions/66171881/how-to-read-nested-txt-file-from-spring-integration-file The `FileInboundChannelAdapterSpec` can be configured with an external `DirectoryScanner`, but it sometimes becomes burden for end-users to extract a scanner bean and configure it with all the required file filters * Expose a `recursive(boolean)` option for better end-user experience * Rework `FileTests` for JUnit 5 * Mention a new option in the docs * * Restore accidentally removed code * Restore special symbols in the `FileTests` * Fix language in the docs according review
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2020 the original author or authors.
|
||||
* Copyright 2016-2021 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.
|
||||
@@ -22,13 +22,14 @@ import java.util.Comparator;
|
||||
import java.util.Map;
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.integration.dsl.ComponentsRegistration;
|
||||
import org.springframework.integration.dsl.MessageSourceSpec;
|
||||
import org.springframework.integration.expression.FunctionExpression;
|
||||
import org.springframework.integration.file.DirectoryScanner;
|
||||
import org.springframework.integration.file.FileLocker;
|
||||
import org.springframework.integration.file.FileReadingMessageSource;
|
||||
import org.springframework.integration.file.RecursiveDirectoryScanner;
|
||||
import org.springframework.integration.file.config.FileListFilterFactoryBean;
|
||||
import org.springframework.integration.file.filters.ExpressionFileListFilter;
|
||||
import org.springframework.integration.file.filters.FileListFilter;
|
||||
@@ -75,6 +76,22 @@ public class FileInboundChannelAdapterSpec
|
||||
return _this();
|
||||
}
|
||||
|
||||
/**
|
||||
* A convenient flag to determine if target message source should use a
|
||||
* {@link RecursiveDirectoryScanner} or stay with a default one.
|
||||
* @param recursive to set or not a {@link RecursiveDirectoryScanner}.
|
||||
* @return the spec.
|
||||
* @see org.springframework.integration.file.RecursiveDirectoryScanner
|
||||
* @since 5.5
|
||||
*/
|
||||
public FileInboundChannelAdapterSpec recursive(boolean recursive) {
|
||||
if (recursive) {
|
||||
new DirectFieldAccessor(this.target).setPropertyValue("scanner", new RecursiveDirectoryScanner());
|
||||
}
|
||||
return _this();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Specify a custom scanner.
|
||||
* @param scanner the scanner.
|
||||
@@ -90,7 +107,7 @@ public class FileInboundChannelAdapterSpec
|
||||
/**
|
||||
* Specify whether to create the source directory automatically if it does
|
||||
* not yet exist upon initialization. By default, this value is
|
||||
* <em>true</em>. If set to <em>false</em> and the
|
||||
* {@code true}. If set to {@code false} and the
|
||||
* source directory does not exist, an Exception will be thrown upon
|
||||
* initialization.
|
||||
* @param autoCreateDirectory the autoCreateDirectory.
|
||||
@@ -262,12 +279,7 @@ public class FileInboundChannelAdapterSpec
|
||||
@Override
|
||||
public Map<Object, String> getComponentsToRegister() {
|
||||
if (this.scanner == null || this.filtersSet) {
|
||||
try {
|
||||
this.target.setFilter(this.fileListFilterFactoryBean.getObject());
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new BeanCreationException("The bean for the [" + this + "] can not be instantiated.", e);
|
||||
}
|
||||
this.target.setFilter(this.fileListFilterFactoryBean.getObject());
|
||||
}
|
||||
|
||||
if (this.expressionFileListFilter != null) {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2019 the original author or authors.
|
||||
* Copyright 2016-2021 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.
|
||||
@@ -17,7 +17,7 @@
|
||||
package org.springframework.integration.file.dsl;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.fail;
|
||||
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
@@ -31,10 +31,8 @@ import java.util.TreeSet;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.TemporaryFolder;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
|
||||
import org.springframework.aop.TargetSource;
|
||||
import org.springframework.aop.framework.Advised;
|
||||
@@ -60,10 +58,10 @@ import org.springframework.integration.dsl.MessageChannels;
|
||||
import org.springframework.integration.dsl.Pollers;
|
||||
import org.springframework.integration.dsl.StandardIntegrationFlow;
|
||||
import org.springframework.integration.expression.FunctionExpression;
|
||||
import org.springframework.integration.file.DefaultDirectoryScanner;
|
||||
import org.springframework.integration.file.DefaultFileNameGenerator;
|
||||
import org.springframework.integration.file.FileHeaders;
|
||||
import org.springframework.integration.file.FileReadingMessageSource;
|
||||
import org.springframework.integration.file.RecursiveDirectoryScanner;
|
||||
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
|
||||
import org.springframework.integration.file.filters.ChainFileListFilter;
|
||||
import org.springframework.integration.file.filters.ExpressionFileListFilter;
|
||||
@@ -82,7 +80,7 @@ import org.springframework.messaging.support.ErrorMessage;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
import org.springframework.util.FileCopyUtils;
|
||||
|
||||
/**
|
||||
@@ -90,12 +88,12 @@ import org.springframework.util.FileCopyUtils;
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringJUnitConfig
|
||||
@DirtiesContext
|
||||
public class FileTests {
|
||||
|
||||
@ClassRule
|
||||
public static final TemporaryFolder tmpDir = new TemporaryFolder();
|
||||
@TempDir
|
||||
static File tmpDir;
|
||||
|
||||
@Autowired
|
||||
private ListableBeanFactory beanFactory;
|
||||
@@ -144,14 +142,9 @@ public class FileTests {
|
||||
@Test
|
||||
public void testFileHandler() throws Exception {
|
||||
Message<?> message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build();
|
||||
try {
|
||||
this.fileFlow1Input.send(message);
|
||||
fail("NullPointerException expected");
|
||||
}
|
||||
catch (Exception e) {
|
||||
assertThat(e).isInstanceOf(MessageHandlingException.class);
|
||||
assertThat(e.getCause()).isInstanceOf(NullPointerException.class);
|
||||
}
|
||||
assertThatExceptionOfType(MessageHandlingException.class)
|
||||
.isThrownBy(() -> this.fileFlow1Input.send(message))
|
||||
.withCauseInstanceOf(NullPointerException.class);
|
||||
DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
|
||||
fileNameGenerator.setBeanFactory(this.beanFactory);
|
||||
Object targetFileWritingMessageHandler = this.fileWritingMessageHandler;
|
||||
@@ -167,7 +160,7 @@ public class FileTests {
|
||||
dfa.setPropertyValue("fileNameGenerator", fileNameGenerator);
|
||||
this.fileFlow1Input.send(message);
|
||||
|
||||
assertThat(new File(tmpDir.getRoot(), "foo").exists()).isTrue();
|
||||
assertThat(new File(tmpDir, "foo").exists()).isTrue();
|
||||
|
||||
this.fileTriggerFlowInput.send(new GenericMessage<>("trigger"));
|
||||
assertThat(this.flushPredicateCalled.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
@@ -175,10 +168,12 @@ public class FileTests {
|
||||
|
||||
@Test
|
||||
public void testMessageProducerFlow() throws Exception {
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir.getRoot(), "TailTest"));
|
||||
File tailTestFile = new File(tmpDir, "TailTest");
|
||||
FileOutputStream file = new FileOutputStream(tailTestFile);
|
||||
for (int i = 0; i < 50; i++) {
|
||||
file.write((i + "\n").getBytes());
|
||||
}
|
||||
file.close();
|
||||
this.tailer.start();
|
||||
for (int i = 0; i < 50; i++) {
|
||||
Message<?> message = this.tailChannel.receive(5000);
|
||||
@@ -188,7 +183,10 @@ public class FileTests {
|
||||
assertThat(this.tailChannel.receive(1)).isNull();
|
||||
|
||||
this.controlBus.send("@tailer.stop()");
|
||||
file.close();
|
||||
|
||||
while (!tailTestFile.delete()) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
}
|
||||
|
||||
@Autowired
|
||||
@@ -203,7 +201,7 @@ public class FileTests {
|
||||
if (even) {
|
||||
evens.add(i);
|
||||
}
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir.getRoot(), i + extension));
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir, i + extension));
|
||||
file.write(("" + i).getBytes());
|
||||
file.flush();
|
||||
file.close();
|
||||
@@ -218,7 +216,7 @@ public class FileTests {
|
||||
assertThat(result.size()).isEqualTo(25);
|
||||
result.forEach(s -> assertThat(evens.contains(Integer.parseInt(s))).isTrue());
|
||||
|
||||
new File(tmpDir.getRoot(), "a.sitest").createNewFile();
|
||||
new File(tmpDir, "a.sitest").createNewFile();
|
||||
Message<?> receive = this.filePollingErrorChannel.receive(60000);
|
||||
assertThat(receive).isNotNull();
|
||||
assertThat(receive).isInstanceOf(ErrorMessage.class);
|
||||
@@ -247,7 +245,7 @@ public class FileTests {
|
||||
|
||||
@Test
|
||||
public void testFileSplitterFlow() throws Exception {
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir.getRoot(), "foo.tmp"));
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir, "foo.tmp"));
|
||||
file.write(("HelloWorld\näöüß").getBytes(Charset.defaultCharset()));
|
||||
file.flush();
|
||||
file.close();
|
||||
@@ -277,13 +275,13 @@ public class FileTests {
|
||||
|
||||
@Test
|
||||
public void testDynamicFileFlows() throws Exception {
|
||||
File newFolder1 = tmpDir.newFolder();
|
||||
File newFolder1 = java.nio.file.Files.createTempDirectory(tmpDir.toPath(), "junit").toFile();
|
||||
FileOutputStream file = new FileOutputStream(new File(newFolder1, "foo"));
|
||||
file.write(("foo").getBytes());
|
||||
file.flush();
|
||||
file.close();
|
||||
|
||||
File newFolder2 = tmpDir.newFolder();
|
||||
File newFolder2 = java.nio.file.Files.createTempDirectory(tmpDir.toPath(), "junit").toFile();
|
||||
file = new FileOutputStream(new File(newFolder2, "bar"));
|
||||
file.write(("bar").getBytes());
|
||||
file.flush();
|
||||
@@ -299,7 +297,12 @@ public class FileTests {
|
||||
assertThat(receive).isNotNull();
|
||||
payloads.add((String) receive.getPayload());
|
||||
|
||||
assertThat(payloads.toArray()).isEqualTo(new String[] { "bar", "foo" });
|
||||
assertThat(payloads.toArray()).isEqualTo(new String[]{ "bar", "foo" });
|
||||
|
||||
assertThat(TestUtils.getPropertyValue(
|
||||
this.beanFactory.getBean(newFolder1.getName() + ".adapter.source"),
|
||||
"scanner"))
|
||||
.isInstanceOf(RecursiveDirectoryScanner.class);
|
||||
}
|
||||
|
||||
@MessagingGateway(defaultRequestChannel = "controlBus.input")
|
||||
@@ -333,7 +336,7 @@ public class FileTests {
|
||||
@Bean
|
||||
public IntegrationFlow fileFlow1() {
|
||||
return IntegrationFlows.from("fileFlow1Input")
|
||||
.handle(Files.outboundAdapter("'file://" + tmpDir.getRoot().getAbsolutePath() + '\'')
|
||||
.handle(Files.outboundAdapter("'file://" + tmpDir.getAbsolutePath() + '\'')
|
||||
.fileNameGenerator(message -> null)
|
||||
.fileExistsMode(FileExistsMode.APPEND_NO_FLUSH)
|
||||
.flushInterval(60000)
|
||||
@@ -349,7 +352,7 @@ public class FileTests {
|
||||
@Bean
|
||||
public IntegrationFlow tailFlow() {
|
||||
return IntegrationFlows
|
||||
.from(Files.tailAdapter(new File(tmpDir.getRoot(), "TailTest"))
|
||||
.from(Files.tailAdapter(new File(tmpDir, "TailTest"))
|
||||
.delay(500)
|
||||
.end(false)
|
||||
.id("tailer")
|
||||
@@ -362,7 +365,7 @@ public class FileTests {
|
||||
@Bean
|
||||
public IntegrationFlow fileReadingFlow() {
|
||||
return IntegrationFlows
|
||||
.from(Files.inboundAdapter(tmpDir.getRoot())
|
||||
.from(Files.inboundAdapter(tmpDir)
|
||||
.patternFilter("*.sitest")
|
||||
.useWatchService(true)
|
||||
.watchEvents(FileReadingMessageSource.WatchEventType.CREATE,
|
||||
@@ -387,7 +390,7 @@ public class FileTests {
|
||||
public IntegrationFlow fileWritingFlow() {
|
||||
return IntegrationFlows.from("fileWritingInput")
|
||||
.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.write")
|
||||
.header("directory", new File(tmpDir.getRoot(), "fileWritingFlow")))
|
||||
.header("directory", new File(tmpDir, "fileWritingFlow")))
|
||||
.handle(Files.outboundGateway(m -> m.getHeaders().get("directory"))
|
||||
.preserveTimestamp(true)
|
||||
.chmod(0777))
|
||||
@@ -403,7 +406,7 @@ public class FileTests {
|
||||
fileExpressionFileListFilter.setBeanFactory(beanFactory);
|
||||
|
||||
return IntegrationFlows
|
||||
.from(Files.inboundAdapter(tmpDir.getRoot())
|
||||
.from(Files.inboundAdapter(tmpDir)
|
||||
.filter(new ChainFileListFilter<File>()
|
||||
.addFilter(new AcceptOnceFileListFilter<>())
|
||||
.addFilter(fileExpressionFileListFilter)),
|
||||
@@ -437,8 +440,7 @@ public class FileTests {
|
||||
void pollDirectories(File... directories) {
|
||||
for (File directory : directories) {
|
||||
StandardIntegrationFlow integrationFlow = IntegrationFlows
|
||||
.from(Files.inboundAdapter(directory)
|
||||
.scanner(new DefaultDirectoryScanner()),
|
||||
.from(Files.inboundAdapter(directory).recursive(true),
|
||||
e -> e.poller(p -> p.fixedDelay(1000))
|
||||
.id(directory.getName() + ".adapter"))
|
||||
.transform(Files.toStringTransformer(),
|
||||
|
||||
@@ -6,8 +6,8 @@ Spring Integration's file support extends the Spring Integration core with a ded
|
||||
You need to include this dependency into your project:
|
||||
|
||||
====
|
||||
[source, xml, subs="normal", role="primary"]
|
||||
.Maven
|
||||
[source, xml, subs="normal"]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
@@ -15,9 +15,8 @@ You need to include this dependency into your project:
|
||||
<version>{project-version}</version>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
[source, groovy, subs="normal", role="secondary"]
|
||||
.Gradle
|
||||
[source, groovy, subs="normal"]
|
||||
----
|
||||
compile "org.springframework.integration:spring-integration-file:{project-version}"
|
||||
----
|
||||
@@ -216,6 +215,8 @@ All other sub-directories inclusions and exclusions are based on the target `Fil
|
||||
For example, the `SimplePatternFileListFilter` filters out directories by default.
|
||||
See https://docs.spring.io/spring-integration/api/org/springframework/integration/file/filters/AbstractDirectoryAwareFileListFilter.html[`AbstractDirectoryAwareFileListFilter`] and its implementations for more information.
|
||||
|
||||
NOTE: Starting with version 5.5, the `FileInboundChannelAdapterSpec` of the Java DSL has a convenient `recursive(boolean)` option to use a `RecursiveDirectoryScanner` in the target `FileReadingMessageSource` instead of the default one.
|
||||
|
||||
[[file-namespace-support]]
|
||||
==== Namespace Support
|
||||
|
||||
|
||||
@@ -59,6 +59,8 @@ In addition, `forRecursion=true` causes the full path to files to be used as the
|
||||
IMPORTANT: This means that existing keys in a persistent metadata store will not be found for files beneath the top level directory.
|
||||
For this reason, the property is `false` by default; this may change in a future release.
|
||||
|
||||
The `FileInboundChannelAdapterSpec` has now a convenient `recursive(boolean)` option instead of requiring an explicit reference to the `RecursiveDirectoryScanner`.
|
||||
|
||||
[[x5.5-mongodb]]
|
||||
==== MongoDb Changes
|
||||
|
||||
|
||||
Reference in New Issue
Block a user