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.

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;
 	}
 
-
 }

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<FileSplitterSpec, FileSplitter> {
 
@@ -40,6 +43,8 @@ public class FileSplitterSpec extends MessageHandlerSpec<FileSplitterSpec, FileS
 
 	private boolean applySequence;
 
+	private String firstLineHeaderName;
+
 	FileSplitterSpec() {
 		this(true);
 	}
@@ -79,7 +84,6 @@ public class FileSplitterSpec extends MessageHandlerSpec<FileSplitterSpec, FileS
 	 * {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker}s
 	 * Defaults to {@code false}.
 	 * @return the FileSplitterSpec
-	 * @see FileSplitter
 	 */
 	public FileSplitterSpec markers() {
 		return markers(false);
@@ -92,7 +96,6 @@ public class FileSplitterSpec extends MessageHandlerSpec<FileSplitterSpec, FileS
 	 * Defaults to {@code false} for markers and {@code false} for markersJson.
 	 * @param asJson the asJson flag to use.
 	 * @return the FileSplitterSpec
-	 * @see FileSplitter
 	 */
 	public FileSplitterSpec markers(boolean asJson) {
 		this.markers = true;
@@ -113,12 +116,25 @@ public class FileSplitterSpec extends MessageHandlerSpec<FileSplitterSpec, FileS
 		return this;
 	}
 
+	/**
+	 * 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.
+	 * @return the FileSplitterSpec
+	 */
+	public FileSplitterSpec firstLineAsHeader(String firstLineHeaderName) {
+		this.firstLineHeaderName = firstLineHeaderName;
+		return this;
+	}
 
 	@Override
 	protected FileSplitter doGet() {
 		FileSplitter fileSplitter = new FileSplitter(this.iterator, this.markers, this.markersJson);
 		fileSplitter.setApplySequence(this.applySequence);
 		fileSplitter.setCharset(this.charset);
+		if (StringUtils.hasText(this.firstLineHeaderName)) {
+			fileSplitter.setFirstLineAsHeader(this.firstLineHeaderName);
+		}
 		return fileSplitter;
 	}
 

spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.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.
@@ -47,20 +47,28 @@ import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
- * The {@link AbstractMessageSplitter} implementation to split the {@link File}
- * Message payload to lines.
+ * The {@link AbstractMessageSplitter} implementation to split the {@link File} Message
+ * payload to lines.
  * <p>
- * 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.
  * <p>
- *  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.
+ * <p>
+ * 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<Object> iterator = new Iterator<Object>() {
 
 			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<String> messageBuilder =
+							getMessageBuilderFactory()
+									.withPayload(line);
+
+					if (firstLineAsHeader != null) {
+						messageBuilder.setHeader(FileSplitter.this.firstLineHeaderName, firstLineAsHeader);
+					}
+
+					return messageBuilder;
 				}
 				else {
 					this.done = true;

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.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="first-line-as-header" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						The header name for the first line to be carried as a header in the messages emitted for the
+						remaining lines.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="requires-reply" use="optional" default="false">
 				<xsd:annotation>
 					<xsd:documentation>

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml

@@ -1,26 +1,27 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:int-file="http://www.springframework.org/schema/integration/file"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-file="http://www.springframework.org/schema/integration/file"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
 
-	<int:channel id="out" />
+	<int:channel id="out"/>
 
 	<int-file:splitter id="fullBoat"
-		iterator="false"
-		markers="true"
-		markers-json="true"
-		apply-sequence="true"
-		requires-reply="true"
-		charset="UTF-8"
-		input-channel="in"
-		output-channel="out"
-		send-timeout="5"
-		auto-startup="false"
-		order="2"
-		phase="1" />
+					   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"/>
 
 </beans>

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"));

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")

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 `<publish-subscribe-channel/>`.
+<13> Set the order of this endpoint if the `input-channel` is a `<publish-subscribe-channel/>`.
 
-<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 <<header-enricher, Header Enricher>> 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<File>()
+                        .addFilter(new AcceptOnceFileListFilter<>())
+                        .addFilter(new ExpressionFileListFilter<>(
+                             new FunctionExpression<File>(f -> "foo.tmp".equals(f.getName()))))))
+            .split(Files.splitter()
+                     .markers()
+                     .charset(StandardCharsets.US_ASCII)
+                     .firstLineAsHeader("fileHeader")
+                     .applySequence(true))
+            .channel(c -> c.queue("fileSplittingResultChannel"))
+            .get();
+    }
+
+}
+----

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 <<file-incomplete>> 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 <<files>> for more information.
 
 ==== (S)FTP Changes