From 8f0fa3468fcf7298d28fae0f40b2cbd94257afc3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 20 Sep 2016 15:09:36 -0400 Subject: [PATCH] INT-832: Add File Relative Path to Message Headers JIRA: https://jira.spring.io/browse/INT-832 When we scan directory recursively for files (e.g. `WatchServiceDirectoryScanner`), it can be useful to get access to the relative path from the `Message`, e.g. on the `FileWritingMessageHandler` side to restore the original structure * Add a `FileHeaders.FILENAME` into the outbound `Message` from the `FileReadingMessageSource` * The result of that header is like a removal of leading `this.directory.getAbsolutePath()` in the target `File.getAbsolutePath()`. In case of not recursion we get just only regular file name. * Introduce `FileHeaders.RELATIVE_PATH` * Populated `FileHeaders.RELATIVE_PATH`, `FileHeaders.FILENAME`, `FileHeaders.ORIGINAL_FILE` in the `FileReadingMessageSource` * File `FileTailingMessageProducerSupport` to populate `FileHeaders` properly * Introduce ctor for the `LastModifiedFileListFilter` for better Java configuration experience * Add docs for changes Doc Polishing --- .../integration/file/FileHeaders.java | 3 +++ .../file/FileReadingMessageSource.java | 9 ++++++- .../filters/LastModifiedFileListFilter.java | 14 +++++++++++ .../FileTailingMessageProducerSupport.java | 4 +++- ...eReadingMessageSourceIntegrationTests.java | 7 +++++- .../file/FileReadingMessageSourceTests.java | 21 ++++++++++------ .../WatchServiceDirectoryScannerTests.java | 19 ++++++++++++++- .../tail/FileTailingMessageProducerTests.java | 4 ++++ ...oundRemoteFileSystemSynchronizerTests.java | 11 +++++---- .../RedisInboundChannelAdapterTests.java | 6 ++--- src/reference/asciidoc/file.adoc | 24 ++++++++++++++++++- src/reference/asciidoc/whats-new.adoc | 5 ++++ 12 files changed, 108 insertions(+), 19 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index ec1ac7e973..36bee26abb 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java @@ -22,6 +22,7 @@ package org.springframework.integration.file; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ public abstract class FileHeaders { @@ -29,6 +30,8 @@ public abstract class FileHeaders { public static final String FILENAME = PREFIX + "name"; + public static final String RELATIVE_PATH = PREFIX + "relativePath"; + public static final String ORIGINAL_FILE = PREFIX + "originalFile"; public static final String REMOTE_DIRECTORY = PREFIX + "remoteDirectory"; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 1b22eaff57..fda2b06aad 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -39,6 +39,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Matcher; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -361,7 +362,13 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } if (file != null) { - message = this.getMessageBuilderFactory().withPayload(file).build(); + message = getMessageBuilderFactory().withPayload(file) + .setHeader(FileHeaders.RELATIVE_PATH, file.getAbsolutePath() + .replaceFirst(Matcher.quoteReplacement(this.directory.getAbsolutePath() + File.separator), + "")) + .setHeader(FileHeaders.FILENAME, file.getName()) + .setHeader(FileHeaders.ORIGINAL_FILE, file) + .build(); if (logger.isInfoEnabled()) { logger.info("Created message: [" + message + "]"); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java index 130c8bcbc0..8dfa08c8b8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java @@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit; * The resolution is done in seconds. * * @author Gary Russell + * @author Artem Bilan * @since 4.2 * */ @@ -42,6 +43,19 @@ public class LastModifiedFileListFilter implements FileListFilter { return this.age; } + public LastModifiedFileListFilter() { + } + + /** + * Construct a {@link LastModifiedFileListFilter} instance with provided {@link #age}. + * Defaults to 60 seconds. + * @param age the age in seconds. + * @since 5.0 + */ + public LastModifiedFileListFilter(long age) { + this.age = age; + } + /** * Set the age that files have to be before being passed by this filter. * If {@link File#lastModified()} plus age is greater than the current time, the file diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java index 90647f2eea..fc03ee3cff 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java @@ -32,6 +32,7 @@ import org.springframework.util.Assert; * Base class for file tailing inbound adapters. * * @author Gary Russell + * @author Artem Bilan * @since 3.0 * */ @@ -101,7 +102,8 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS protected void send(String line) { Message message = this.getMessageBuilderFactory().withPayload(line) - .setHeader(FileHeaders.FILENAME, this.file.getAbsolutePath()) + .setHeader(FileHeaders.FILENAME, this.file.getName()) + .setHeader(FileHeaders.ORIGINAL_FILE, this.file) .build(); super.sendMessage(message); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java index 485db76dfc..8325ec5e63 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java @@ -134,7 +134,12 @@ public class FileReadingMessageSourceIntegrationTests { public void inputDirExhausted() throws Exception { assertNotNull(pollableFileSource.receive()); assertNotNull(pollableFileSource.receive()); - assertNotNull(pollableFileSource.receive()); + Message receive = pollableFileSource.receive(); + assertNotNull(receive); + File payload = receive.getPayload(); + assertEquals(payload, receive.getHeaders().get(FileHeaders.ORIGINAL_FILE)); + assertEquals(payload.getName(), receive.getHeaders().get(FileHeaders.FILENAME)); + assertEquals(payload.getName(), receive.getHeaders().get(FileHeaders.RELATIVE_PATH)); assertNull(pollableFileSource.receive()); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index fcb1f1a63a..7094a2e792 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -44,6 +44,7 @@ import org.springframework.messaging.Message; * @author Iwein Fuld * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell */ @RunWith(MockitoJUnitRunner.class) public class FileReadingMessageSourceTests { @@ -66,6 +67,8 @@ public class FileReadingMessageSourceTests { when(inputDirectoryMock.isDirectory()).thenReturn(true); when(inputDirectoryMock.exists()).thenReturn(true); when(inputDirectoryMock.canRead()).thenReturn(true); + when(inputDirectoryMock.getAbsolutePath()).thenReturn("foo/bar"); + when(fileMock.getAbsolutePath()).thenReturn("foo/bar/fileMock"); when(locker.lock(isA(File.class))).thenReturn(true); } @@ -80,13 +83,13 @@ public class FileReadingMessageSourceTests { @Test public void straightProcess() throws Exception { - when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); assertThat(source.receive().getPayload(), is(fileMock)); } @Test public void requeueOnFailure() throws Exception { - when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); Message received = source.receive(); assertNotNull(received); source.onFailure(received); @@ -97,7 +100,8 @@ public class FileReadingMessageSourceTests { @Test public void scanEachPoll() throws Exception { File anotherFileMock = mock(File.class); - when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock, anotherFileMock}); + when(anotherFileMock.getAbsolutePath()).thenReturn("foo/bar/anotherFileMock"); + when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock, anotherFileMock }); source.setScanEachPoll(true); assertNotNull(source.receive()); assertNotNull(source.receive()); @@ -107,7 +111,7 @@ public class FileReadingMessageSourceTests { @Test public void noDuplication() throws Exception { - when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); Message received = source.receive(); assertNotNull(received); assertEquals(fileMock, received.getPayload()); @@ -122,7 +126,7 @@ public class FileReadingMessageSourceTests { @Test public void lockIsAcquired() throws IOException { - when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); Message received = source.receive(); assertNotNull(received); assertEquals(fileMock, received.getPayload()); @@ -131,7 +135,7 @@ public class FileReadingMessageSourceTests { @Test public void lockedFilesAreIgnored() throws IOException { - when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); when(locker.lock(fileMock)).thenReturn(false); Message received = source.receive(); assertNull(received); @@ -141,8 +145,11 @@ public class FileReadingMessageSourceTests { @Test public void orderedReception() throws Exception { File file1 = mock(File.class); + when(file1.getAbsolutePath()).thenReturn("foo/bar/file1"); File file2 = mock(File.class); + when(file2.getAbsolutePath()).thenReturn("foo/bar/file2"); File file3 = mock(File.class); + when(file3.getAbsolutePath()).thenReturn("foo/bar/file3"); // record the comparator to reverse order the files when(comparator.compare(file1, file2)).thenReturn(1); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java index 12c9926aaa..2d20c92940 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java @@ -17,7 +17,9 @@ package org.springframework.integration.file; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; @@ -39,6 +41,8 @@ import org.junit.rules.TemporaryFolder; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter; import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; /** * @author Gary Russell @@ -71,7 +75,7 @@ public class WatchServiceDirectoryScannerTests { } @Test - public void testInitialAndAddMoreThenRemove() throws Exception { + public void testWatchServiceDirectoryScanner() throws Exception { FileReadingMessageSource fileReadingMessageSource = new FileReadingMessageSource(); fileReadingMessageSource.setDirectory(folder.getRoot()); fileReadingMessageSource.setUseWatchService(true); @@ -188,6 +192,19 @@ public class WatchServiceDirectoryScannerTests { assertTrue(removeFileLatch.await(10, TimeUnit.SECONDS)); + File baz3 = File.createTempFile("baz3", ".txt", baz); + + n = 0; + Message fileMessage = null; + while (n++ < 300 && (fileMessage = fileReadingMessageSource.receive()) == null) { + Thread.sleep(100); + } + + assertNotNull(fileMessage); + assertEquals(baz3, fileMessage.getPayload()); + assertThat(fileMessage.getHeaders().get(FileHeaders.RELATIVE_PATH, String.class), + startsWith(TestUtils.applySystemFileSeparator("foo/baz/"))); + fileReadingMessageSource.stop(); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java index b0aa31468f..bee29e7846 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java @@ -37,12 +37,14 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.tail.FileTailingMessageProducerSupport.FileTailingEvent; import org.springframework.messaging.Message; /** * @author Gary Russell * @author Gavin Gray + * @author Artem Bilan * @since 3.0 * */ @@ -169,6 +171,8 @@ public class FileTailingMessageProducerTests { Message message = outputChannel.receive(10000); assertNotNull("expected a non-null message", message); assertEquals("hello" + i, message.getPayload()); + assertEquals(file, message.getHeaders().get(FileHeaders.ORIGINAL_FILE)); + assertEquals(file.getName(), message.getHeaders().get(FileHeaders.FILENAME)); } } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java index 65b729760b..11a05ac1da 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java @@ -48,6 +48,7 @@ import org.springframework.expression.Expression; import org.springframework.expression.ExpressionParser; import org.springframework.expression.spel.SpelParserConfiguration; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.CompositeFileListFilter; import org.springframework.integration.file.filters.FileListFilter; @@ -85,8 +86,8 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { @Test public void testCopyFileToLocalDir() throws Exception { - File localDirectoy = new File("test"); - assertFalse(localDirectoy.exists()); + File localDirectory = new File("test"); + assertFalse(localDirectory.exists()); TestFtpSessionFactory ftpSessionFactory = new TestFtpSessionFactory(); ftpSessionFactory.setUsername("kermit"); @@ -118,7 +119,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { ms.setAutoCreateLocalDirectory(true); - ms.setLocalDirectory(localDirectoy); + ms.setLocalDirectory(localDirectory); ms.setBeanFactory(mock(BeanFactory.class)); CompositeFileListFilter localFileListFilter = new CompositeFileListFilter(); localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.TEST\\.a$")); @@ -132,6 +133,8 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { // The test remote files are created with the current timestamp + 1 day. assertThat(atestFile.getPayload().lastModified(), Matchers.greaterThan(System.currentTimeMillis())); + assertEquals("A.TEST.a", atestFile.getHeaders().get(FileHeaders.FILENAME)); + Message btestFile = ms.receive(); assertNotNull(btestFile); assertEquals("B.TEST.a", btestFile.getPayload().getName()); @@ -142,7 +145,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { assertNull(nothing); // two times because on the third receive (above) the internal queue will be empty, so it will attempt - verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy, Integer.MIN_VALUE); + verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectory, Integer.MIN_VALUE); assertTrue(new File("test/A.TEST.a").exists()); assertTrue(new File("test/B.TEST.a").exists()); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java index 735eeed762..b0678fcbc2 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java @@ -78,7 +78,7 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests { } int counter = 0; for (int i = 0; i < numToTest; i++) { - Message message = channel.receive(5000); + Message message = channel.receive(10000); if (message == null) { throw new RuntimeException("Failed to receive message # " + i + " iteration " + iteration); } @@ -96,7 +96,7 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests { adapter.afterPropertiesSet(); adapter.start(); - RedisTemplate template = new RedisTemplate(); + RedisTemplate template = new RedisTemplate<>(); template.setConnectionFactory(connectionFactory); template.setEnableDefaultSerializer(false); template.afterPropertiesSet(); @@ -111,7 +111,7 @@ public class RedisInboundChannelAdapterTests extends RedisAvailableTests { counter = 0; for (int i = 0; i < numToTest; i++) { - Message message = channel.receive(5000); + Message message = channel.receive(10000); if (message == null) { throw new RuntimeException("Failed to receive message # " + i + " iteration " + iteration); } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 38f1db063c..edc1dad53c 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -24,7 +24,8 @@ This is an implementation of `MessageSource` that creates messages from a file s p:directory="${input.directory}"/> ---- -To prevent creating messages for certain files, you may supply a `FileListFilter`. By default the following 2 filters are used: +To prevent creating messages for certain files, you may supply a `FileListFilter`. +By default the following 2 filters are used: * `IgnoreHiddenFileListFilter` * `AcceptOnceFileListFilter` @@ -105,6 +106,19 @@ to, say, network glitches. ---- +*Message Headers* + +Starting with _version 5.0_ the `FileReadingMessageSource`, in addition to the `payload` as a polled `File`, populates these headers to the outbound `Message`: + +- `FileHeaders.FILENAME` - the `File.getName()` of the file to send. +Can be used for subsequent rename or copy logic; +- `FileHeaders.ORIGINAL_FILE` - the `File` object itself. +Typically this header is populated automatically by Framework components, like <> or <>, when we lose the original `File` object. +But for consistency and convenience with any other custom use-cases this header can be useful to get access to the original file; +- `FileHeaders.RELATIVE_PATH` - a new header introduced to represent the part of file path relative to the root directory for the scan. +This header can be useful when the requirement is to restore a source directory hierarchy in the other places. +For this purpose the `DefaultFileNameGenerator` (<>) can be configured to use this header. + *Directory scanning and polling* The `FileReadingMessageSource` doesn't produce messages for files from the directory immediately. @@ -399,6 +413,14 @@ This sequence of events might occur, for example, when a file is rotated. NOTE: Not all platforms supporting a `tail` command provide these status messages. +Messages emitted from these endpoints have the following headers: + +- `FileHeaders.ORIGINAL_FILE` - the `File` object +- `FileHeaders.FILENAME` - the file name (`File.getName()`) + +NOTE: In versions prior to _5.0_, the `FileHeaders.FILENAME` header contained a string representation of the file's absolute path. +You can now obtain that by calling `getAbsolutePath()` on the original file header. + Example configurations: [source,xml] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 23e49b1cd6..40b3b8eb56 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -37,6 +37,11 @@ See <> for more information. Some inconsistencies with rendering IMAP mail content have been resolved. See <> for more information. +==== File Changes + +The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to prepresent relative path in the `FileReadingMessageSource`. +See <> for more information. + ==== (S)FTP Changes The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.