diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java index 5e8901b898..e8f32a63bb 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java @@ -26,6 +26,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa */ public class FileNamespaceHandler extends AbstractIntegrationNamespaceHandler { + @Override public void init() { registerBeanDefinitionParser("inbound-channel-adapter", new FileInboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-channel-adapter", new FileOutboundChannelAdapterParser()); @@ -33,6 +34,7 @@ public class FileNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("file-to-string-transformer", new FileToStringTransformerParser()); registerBeanDefinitionParser("file-to-bytes-transformer", new FileToByteArrayTransformerParser()); registerBeanDefinitionParser("tail-inbound-channel-adapter", new FileTailInboundChannelAdapterParser()); + registerBeanDefinitionParser("splitter", new FileSplitterParser()); } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java new file mode 100644 index 0000000000..02bc1b4fb3 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java @@ -0,0 +1,47 @@ +/* + * Copyright 2015 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 + * + * http://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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractConsumerEndpointParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.file.splitter.FileSplitter; + + +/** + * @author Gary Russell + * @since 4.2 + * + */ +public class FileSplitterParser extends AbstractConsumerEndpointParser { + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FileSplitter.class); + builder.addConstructorArgValue(element.getAttribute("iterator")); + builder.addConstructorArgValue(element.getAttribute("markers")); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + return builder; + } + + +} diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd index 2a7950af1a..a5f6ade39b 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd @@ -547,6 +547,92 @@ Only files matching this regular expression will be picked up by this adapter. + + + + Defines a Splitter that splits text-based files into lines. + + + + + + + + + + + + + + + + + + + + + + + + Specify whether the splitter must return at least one message. This value will be + 'false' by default, but if set to 'true', a ReplyRequiredException will be thrown when + the file has no data. + + + + + + + + + + Set this flag to determine whether sequence related headers are added to messages + from this splitter. When true, existing correlation and sequence related headers + are pushed onto a stack; downstream components, such as aggregators may pop + the stack to revert the existing headers after aggregation. Default is 'true' + unless 'markers' is 'true'. Also see 'iterator'. + + + + + + + + + + Set this flag to determine whether the lines are emitted via an iterator (line at a time) + or the whole file is read into memory and then the lines emitted. When this is 'true', if + 'apply-sequence' is also 'true', the 'sequenceSize' header is set to '0'. Default: 'true'. + + + + + + + + + + Set to 'true' to emit start/end of file marker messages before and after the file data. + Markers are messages with 'FileSplitter.FileMarker' payloads (with 'START' and 'END' values + in the 'mark' property). + Markers might be used when sequentially processing files in a downstream flow where some lines + are filtered. + The 'END' marker includes a line count. + They enable the downstream processing to know when a file has been completely processed. + Default: 'false'. + When 'true', 'apply-sequence' is 'false' by default. + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml new file mode 100644 index 0000000000..44066d9884 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml @@ -0,0 +1,25 @@ + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java new file mode 100644 index 0000000000..68a71fd153 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java @@ -0,0 +1,74 @@ +/* + * Copyright 2015 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 + * + * http://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.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.nio.charset.Charset; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class FileSplitterParserTests { + + @Autowired + private EventDrivenConsumer fullBoat; + + @Autowired + private FileSplitter splitter; + + @Autowired + private MessageChannel in; + + @Autowired + private MessageChannel out; + + @Test + public void testComplete() { + assertFalse(TestUtils.getPropertyValue(this.splitter, "iterator", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(this.splitter, "markers", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(this.splitter, "requiresReply", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(this.splitter, "applySequence", Boolean.class)); + assertEquals(Charset.forName("UTF-8"), TestUtils.getPropertyValue(this.splitter, "charset")); + assertEquals(5L, TestUtils.getPropertyValue(this.splitter, "messagingTemplate.sendTimeout")); + assertEquals(this.out, TestUtils.getPropertyValue(this.splitter, "outputChannel")); + assertEquals(2, TestUtils.getPropertyValue(this.splitter, "order")); + + assertEquals(this.in, TestUtils.getPropertyValue(this.fullBoat, "inputChannel")); + assertFalse(TestUtils.getPropertyValue(this.fullBoat, "autoStartup", Boolean.class)); + assertEquals(1, TestUtils.getPropertyValue(this.fullBoat, "phase")); + } + +} diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index e86558e2f7..28dccdb92c 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -470,3 +470,84 @@ To configure File specific transformers you can use the appropriate elements fro The _delete-files_ option signals to the transformer that it should delete the inbound File after the transformation is complete. This is in no way a replacement for using the`AcceptOnceFileListFilter` when the FileReadingMessageSource is being used in a multi-threaded environment (e.g. Spring Integration in general). + + +[[file-splitter]] +=== File Splitter + +The `FileSplitter` was added in _version 4.1.2_ and namespace support was added in _version 4.2_. +The `FileSplitter` splits text files into individual lines, based on `BufferedReader.readLine()`. +By default, the splitter uses an `Iterator` to emit lines one-at-a-time as they are read from the file. +Setting the `iterator` property to `false` causes it to read all the lines into memory before emitting them as messages. +One use case for this might be if you want to detect I/O errors on the file before sending any messages containing +lines. +However, it is only practical for relatively short files. + +Inbound payloads can be `File`, `String` (a `File` path), `InputStream`, or `Reader`. +Other payload types will be emitted unchanged. + +[source, xml] +---- + + iterator="" <2> + markers="" <3> + apply-sequence="" <4> + requires-reply="" <5> + charset="" <6> + input-channel="" <7> + output-channel="" <8> + send-timeout="" <9> + auto-startup="" <10> + order="" <11> + phase="" /> <12> +---- + +<1> The bean name of the splitter. + +<2> Set to `true` to use an iterator (default); `false` to load the file into memory before sending lines. + +<3> Set to `true` to emit start/end of file marker messages before and after the file data. +Markers are messages with `FileSplitter.FileMarker` payloads (with `START` and `END` values in the `mark` property). +Markers might be used when sequentially processing files in a downstream flow where some lines are filtered. +They enable the downstream processing to know when a file has been completely processed. +The 'END' marker includes a line count. +Default: `false`. +When `true`, `apply-sequence` is `false` by default. + +<4> Set to `false` to disable the inclusion of `sequenceSize` and `sequenceNumber` headers in messages. +Default: `true`, unless `markers` is `true`. +When `true` and `markers` is `true`, the markers are included in the sequencing. +When `true` and `iterator` is `true`, the `sequenceSize` header is set to `0` because the size is unknown. + +<5> Set to `true` to cause a `RequiresReplyException` to be thrown if there are no lines in the file. +Default: `false`. + +<6> Set the charset name to be used when reading the text data into `String` payloads. +Default: platform charset. + +<7> Set the input channel used to send messages to the splitter. + +<8> Set the output channel to which messages will be sent. + +<9> Set the send timeout - only applies if the `output-channel` can block - such as a full `QueueChannel`. + +<10> Set to `false` to disable automatically starting the splitter when the context is refreshed. +Default: `true`. + +<11> Set the order of this endpoint if the `input-channel` is a ``. + +<12> Set the startup phase for the splitter (used when `auto-startup` is `true`). + +*Java Configuration* + +[source, java] +---- +@Splitter(inputChannel="toSplitter") +@Bean +public MessageHandler fileSplitter() { + FileSplitter splitter = new FileSplitter(true, true); + splitter.setApplySequence(true); + splitter.setOutputChannel(outputChannel); + return splitter; +} +---- diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 82b9090247..2613e7d34d 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -177,7 +177,7 @@ Unless your application removes files after processing, the adapter will re-proc Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed. Use the `local-filter` attribute to configure the behavior of the local file system filter. -To solve these particular use cases, you can use a`FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead. +To solve these particular use cases, you can use a `FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead. This filter also stores the accepted file names and modified timestamp in an instance of the`MetadataStore` strategy (<>), and will detect the change in the local file modified time. Since __version 4.1.5__, these filters have a new property `flushOnUpdate` which will cause them to flush the diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 5f1e30a406..7c718e89df 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -263,7 +263,7 @@ Unless your application removes files after processing, the adapter will re-proc Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed. Use the `local-filter` attribute to configure the behavior of the local file system filter. -To solve these particular use cases, you can use a`FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead. +To solve these particular use cases, you can use a `FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead. This filter also stores the accepted file names and modified timestamp in an instance of the`MetadataStore` strategy (<>), and will detect the change in the local file modified time. Since __version 4.1.5__, these filters have a new property `flushOnUpdate` which will cause them to flush the diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index ad6e1d3301..3384553189 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -29,6 +29,12 @@ The `@SecuredChannel` annotation has been introduced, replacing the deprecated ` For more information, see <>. +[[x4.2-file-splitter]] +==== FileSplitter + +The `FileSplitter`, which splits text files into lines, was added in 4.1.2. +It now has full support in the `int-file:` namespace; see <> for more information. + [[x4.2-general]] === General Changes