From d8382e0bd04b4a650213012e5ff0e1900f2cbde0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 26 Jun 2017 16:53:25 -0400 Subject: [PATCH] INT-4297: Add FileSplitter.setFirstLineAsHeader() JIRA: https://jira.spring.io/browse/INT-4297 * Add `FileSplitter.setFirstLineAsHeader()` and extract the first line from the context before `iterator`. * Populate the extracted header to each subsequent line * Don't use `Streams` to read first line * Add XSD and Java DSL option for the `firstLineAsHeader` * Document the feature * Add Java DSL sample for the `Files.slitter()` Add a note about more complex header enrichment Doc Polishing and fix field name. --- .../file/config/FileSplitterParser.java | 6 +- .../file/dsl/FileSplitterSpec.java | 22 +++- .../file/splitter/FileSplitter.java | 67 ++++++++++-- .../config/spring-integration-file-5.0.xsd | 8 ++ .../FileSplitterParserTests-context.xml | 35 +++--- .../file/config/FileSplitterParserTests.java | 6 +- .../file/splitter/FileSplitterTests.java | 58 ++++++++++ src/reference/asciidoc/file.adoc | 100 +++++++++++++----- src/reference/asciidoc/whats-new.adoc | 2 + 9 files changed, 243 insertions(+), 61 deletions(-) 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 index 42d9bd9b3a..20489a30a1 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -27,6 +27,8 @@ import org.springframework.integration.file.splitter.FileSplitter; /** * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 * */ @@ -42,8 +44,8 @@ public class FileSplitterParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "first-line-as-header"); return builder; } - } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java index 4dbe1adf43..4dd0eb4612 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -20,6 +20,7 @@ import java.nio.charset.Charset; import org.springframework.integration.dsl.MessageHandlerSpec; import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.util.StringUtils; /** * The {@link MessageHandlerSpec} for the {@link FileSplitter}. @@ -27,6 +28,8 @@ import org.springframework.integration.file.splitter.FileSplitter; * @author Artem Bilan * * @since 5.0 + * + * @see FileSplitter */ public class FileSplitterSpec extends MessageHandlerSpec { @@ -40,6 +43,8 @@ public class FileSplitterSpec extends MessageHandlerSpec - * With {@code iterator = true} (defaults to {@code true}) this class produces an {@link Iterator} - * to process file lines on demand from {@link Iterator#next}. - * Otherwise a {@link List} of all lines is returned to the to further + * With {@code iterator = true} (defaults to {@code true}) this class produces an + * {@link Iterator} to process file lines on demand from {@link Iterator#next}. Otherwise + * a {@link List} of all lines is returned to the to further * {@link AbstractMessageSplitter#handleRequestMessage} process. *

- * Can accept {@link String} as file path, {@link File}, {@link Reader} or {@link InputStream} - * as payload type. - * All other types are ignored and returned to the {@link AbstractMessageSplitter} as is. + * Can accept {@link String} as file path, {@link File}, {@link Reader} or + * {@link InputStream} as payload type. All other types are ignored and returned to the + * {@link AbstractMessageSplitter} as is. + *

+ * If {@link #setFirstLineAsHeader(String)} is specified, the first line of the content is + * treated as a header and carried as a header with the provided name in the messages + * emitted for the remaining lines. In this case, if markers are enabled, the line count + * in the END marker does not include the header line and, if + * {@link #setApplySequence(boolean) applySequence} is true, the header is not included in + * the sequence. * * @author Artem Bilan * @author Gary Russell + * * @since 4.1.2 */ public class FileSplitter extends AbstractMessageSplitter { @@ -76,6 +84,8 @@ public class FileSplitter extends AbstractMessageSplitter { private Charset charset; + private String firstLineHeaderName; + /** * Construct a splitter where the {@link #splitMessage(Message)} method returns * an iterator and the file is read line-by-line during iteration. @@ -143,6 +153,17 @@ public class FileSplitter extends AbstractMessageSplitter { this.charset = charset; } + /** + * Specify the header name for the first line to be carried as a header in the + * messages emitted for the remaining lines. + * @param firstLineHeaderName the header name to carry first line. + * @since 5.0 + */ + public void setFirstLineAsHeader(String firstLineHeaderName) { + Assert.hasText(firstLineHeaderName, "'firstLineHeaderName' must not be empty"); + this.firstLineHeaderName = firstLineHeaderName; + } + @Override protected Object splitMessage(final Message message) { Object payload = message.getPayload(); @@ -199,7 +220,8 @@ public class FileSplitter extends AbstractMessageSplitter { super.close(); } finally { - Closeable closeableResource = new IntegrationMessageHeaderAccessor(message).getCloseableResource(); + Closeable closeableResource = new IntegrationMessageHeaderAccessor(message) + .getCloseableResource(); if (closeableResource != null) { closeableResource.close(); } @@ -208,6 +230,20 @@ public class FileSplitter extends AbstractMessageSplitter { }; + String firstLineAsHeader; + + if (this.firstLineHeaderName != null) { + try { + firstLineAsHeader = bufferedReader.readLine(); + } + catch (IOException e) { + throw new MessageHandlingException(message, "IOException while reading first line", e); + } + } + else { + firstLineAsHeader = null; + } + Iterator iterator = new Iterator() { boolean markers = FileSplitter.this.markers; @@ -275,7 +311,16 @@ public class FileSplitter extends AbstractMessageSplitter { String line = this.line; this.line = null; this.lineCount++; - return line; + + AbstractIntegrationMessageBuilder messageBuilder = + getMessageBuilderFactory() + .withPayload(line); + + if (firstLineAsHeader != null) { + messageBuilder.setHeader(FileSplitter.this.firstLineHeaderName, firstLineAsHeader); + } + + return messageBuilder; } else { this.done = true; diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd index 154e533020..abf5deff56 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd @@ -712,6 +712,14 @@ Only files matching this regular expression will be picked up by this adapter. + + + + The header name for the first line to be carried as a header in the messages emitted for the + remaining lines. + + + 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 index 5818b86f72..583abb6503 100644 --- 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 @@ -1,26 +1,27 @@ - + + iterator="false" + markers="true" + markers-json="true" + apply-sequence="true" + requires-reply="true" + charset="UTF-8" + first-line-as-header="foo" + input-channel="in" + output-channel="out" + send-timeout="5" + auto-startup="false" + order="2" + phase="1"/> 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 index c388bfb819..aa6712c0a6 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -36,6 +36,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 * */ @@ -67,7 +69,7 @@ public class FileSplitterParserTests { assertEquals(5L, TestUtils.getPropertyValue(this.splitter, "messagingTemplate.sendTimeout")); assertEquals(this.out, TestUtils.getPropertyValue(this.splitter, "outputChannel")); assertEquals(2, TestUtils.getPropertyValue(this.splitter, "order")); - + assertEquals("foo", TestUtils.getPropertyValue(this.splitter, "firstLineHeaderName")); 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/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java index 0a436d04da..4aed0f3fef 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java @@ -307,6 +307,64 @@ public class FileSplitterTests { .verifyComplete(); } + @Test + public void testFirstLineAsHeader() { + QueueChannel outputChannel = new QueueChannel(); + FileSplitter splitter = new FileSplitter(true, true); + splitter.setFirstLineAsHeader("firstLine"); + splitter.setOutputChannel(outputChannel); + splitter.handleMessage(new GenericMessage<>(file)); + Message received = outputChannel.receive(0); + assertNotNull(received); + assertNull(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)); + assertNull(received.getHeaders().get("firstLine")); + assertEquals("START", received.getHeaders().get(FileHeaders.MARKER)); + assertThat(received.getPayload(), instanceOf(FileSplitter.FileMarker.class)); + FileMarker fileMarker = (FileSplitter.FileMarker) received.getPayload(); + assertEquals(FileSplitter.FileMarker.Mark.START, fileMarker.getMark()); + assertEquals(file.getAbsolutePath(), fileMarker.getFilePath()); + received = outputChannel.receive(0); + assertEquals("HelloWorld", received.getHeaders().get("firstLine")); + assertNotNull(received); + received = outputChannel.receive(0); + assertNotNull(received); + assertEquals("END", received.getHeaders().get(FileHeaders.MARKER)); + assertNull(received.getHeaders().get("firstLine")); + assertThat(received.getPayload(), instanceOf(FileSplitter.FileMarker.class)); + fileMarker = (FileSplitter.FileMarker) received.getPayload(); + assertEquals(FileSplitter.FileMarker.Mark.END, fileMarker.getMark()); + assertEquals(file.getAbsolutePath(), fileMarker.getFilePath()); + assertEquals(1, fileMarker.getLineCount()); + } + + @Test + public void testFirstLineAsHeaderOnlyHeader() throws IOException { + QueueChannel outputChannel = new QueueChannel(); + FileSplitter splitter = new FileSplitter(true, true); + splitter.setFirstLineAsHeader("firstLine"); + splitter.setOutputChannel(outputChannel); + File file = File.createTempFile("empty", ".txt"); + splitter.handleMessage(new GenericMessage<>(file)); + Message received = outputChannel.receive(0); + assertNotNull(received); + assertNull(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)); + assertNull(received.getHeaders().get("firstLine")); + assertEquals("START", received.getHeaders().get(FileHeaders.MARKER)); + assertThat(received.getPayload(), instanceOf(FileSplitter.FileMarker.class)); + FileMarker fileMarker = (FileSplitter.FileMarker) received.getPayload(); + assertEquals(FileSplitter.FileMarker.Mark.START, fileMarker.getMark()); + assertEquals(file.getAbsolutePath(), fileMarker.getFilePath()); + received = outputChannel.receive(0); + assertNotNull(received); + assertEquals("END", received.getHeaders().get(FileHeaders.MARKER)); + assertNull(received.getHeaders().get("firstLine")); + assertThat(received.getPayload(), instanceOf(FileSplitter.FileMarker.class)); + fileMarker = (FileSplitter.FileMarker) received.getPayload(); + assertEquals(FileSplitter.FileMarker.Mark.END, fileMarker.getMark()); + assertEquals(file.getAbsolutePath(), fileMarker.getFilePath()); + assertEquals(0, fileMarker.getLineCount()); + } + @Configuration @EnableIntegration @ImportResource("classpath:org/springframework/integration/file/splitter/FileSplitterTests-context.xml") diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 65d93a04f9..30b1b40ea5 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -911,12 +911,13 @@ Other payload types will be emitted unchanged. apply-sequence="" <5> requires-reply="" <6> charset="" <7> - input-channel="" <8> - output-channel="" <9> - send-timeout="" <10> - auto-startup="" <11> - order="" <12> - phase="" /> <13> + first-line-as-header="" <8> + input-channel="" <9> + output-channel="" <10> + send-timeout="" <11> + auto-startup="" <12> + order="" <13> + phase="" /> <14> ---- <1> The bean name of the splitter. @@ -948,32 +949,21 @@ Default: `false`. <7> Set the charset name to be used when reading the text data into `String` payloads. Default: platform charset. -<8> Set the input channel used to send messages to the splitter. +<8> The header name for the first line to be carried as a header in the messages emitted for the remaining lines. +Since _version 5.0_. -<9> Set the output channel to which messages will be sent. +<9> Set the input channel used to send messages to the splitter. -<10> Set the send timeout - only applies if the `output-channel` can block - such as a full `QueueChannel`. +<10> Set the output channel to which messages will be sent. -<11> Set to `false` to disable automatically starting the splitter when the context is refreshed. +<11> Set the send timeout - only applies if the `output-channel` can block - such as a full `QueueChannel`. + +<12> Set to `false` to disable automatically starting the splitter when the context is refreshed. Default: `true`. -<12> Set the order of this endpoint if the `input-channel` is a ``. +<13> Set the order of this endpoint if the `input-channel` is a ``. -<13> 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; -} ----- +<14> Set the startup phase for the splitter (used when `auto-startup` is `true`). The `FileSplitter` will also split any text-based `InputStream` into lines. When used in conjunction with an FTP or SFTP streaming inbound channel adapter, or an FTP or SFTP outbound gateway @@ -990,3 +980,61 @@ public FileSplitter(boolean iterator, boolean markers, boolean markersJson) ---- When `markersJson` is true, the markers will be represented as a JSON string, as long as a suitable JSON processor library, such as Jackson or Boon, is on the classpath. + +Starting with _version 5.0_, the `firstLineAsHeader` option is introduced to specify that the first line of content is a header (such as column names in a CSV file). +The argument passed to this property is the header name under which the first line will be carried as a header in the messages emitted for the remaining lines. +This line is not included in the sequence header (if `applySequence` is true) nor in the `FileMarker.END` `lineCount`. +If file contains only the header line, the file is treated as empty and therefore only `FileMarker` s are emitted during splitting (if markers are enabled, otherwise no messages are emitted). +By default (if no header name is set), the first line is considered data and will be the payload of the first emitted message. + +If you need more complex logic about headers extraction from the file content (not first line, not the whole content of the line, not one header etc.), consider to use <> upfront of the `FileSplitter`. +The lines which have been moved to the headers might be filtered downstream from the normal content process. + +==== Configuring with 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; +} +---- + +==== Configuring with the Java DSL + +The following Spring Boot application provides an example of configuring the inbound adapter using the Java DSL: + +[source, java] +---- +@SpringBootApplication +public class FileSplitterApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(FileSplitterApplication.class) + .web(false) + .run(args); + } + + @Bean + public IntegrationFlow fileSplitterFlow() { + return IntegrationFlows + .from(Files.inboundAdapter(tmpDir.getRoot()) + .filter(new ChainFileListFilter() + .addFilter(new AcceptOnceFileListFilter<>()) + .addFilter(new ExpressionFileListFilter<>( + new FunctionExpression(f -> "foo.tmp".equals(f.getName())))))) + .split(Files.splitter() + .markers() + .charset(StandardCharsets.US_ASCII) + .firstLineAsHeader("fileHeader") + .applySequence(true)) + .channel(c -> c.queue("fileSplittingResultChannel")) + .get(); + } + +} +---- diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 6ca4e47439..531c885f8e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -147,6 +147,8 @@ They also now support setting file permissions on the newly written file. A new `FileSystemMarkerFilePresentFileListFilter` is now available; see <> for more information. +The `FileSplitter` now provides a `firstLineAsHeader` option to carry the first line of content as a header in the messages emitted for the remaining lines. + See <> for more information. ==== (S)FTP Changes