diff --git a/.gitignore b/.gitignore index 11fee3f1a0..354097cc50 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ .pmd .project .settings +.checkstyle bin build build.log diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java index 16d06f07ad..c634025788 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java @@ -39,6 +39,15 @@ public final class JsonObjectMapperProvider { private static final boolean boonPresent = ClassUtils.isPresent("org.boon.json.ObjectMapper", classLoader); + private JsonObjectMapperProvider() { + super(); + } + + /** + * Return an object mapper if available. + * @return the mapper. + * @throws IllegalStateException if an implementation is not available. + */ public static JsonObjectMapper newInstance() { if (JacksonJsonUtils.isJackson2Present()) { return new Jackson2JsonObjectMapper(); @@ -51,4 +60,13 @@ public final class JsonObjectMapperProvider { } } + /** + * Returns true if a supported JSON implementation is on the class path. + * @return true if {@link #newInstance()} will return a mapper. + * @since 4.2.7 + */ + public static boolean jsonAvailable() { + return JacksonJsonUtils.isJackson2Present() || boonPresent; + } + } 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 c88f2a84f5..0132040e24 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 @@ -39,4 +39,9 @@ public abstract class FileHeaders { public static final String RENAME_TO = PREFIX + "renameTo"; + /** + * Record is a file marker (START/END) + */ + public static final String MARKER = PREFIX + "marker"; + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java index d2bd488153..8a85f5b23a 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java @@ -36,10 +36,14 @@ import java.util.NoSuchElementException; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark; import org.springframework.integration.splitter.AbstractMessageSplitter; +import org.springframework.integration.support.json.JsonObjectMapper; +import org.springframework.integration.support.json.JsonObjectMapperProvider; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.util.StringUtils; +import reactor.core.support.Assert; + /** * The {@link AbstractMessageSplitter} implementation to split the {@link File} * Message payload to lines. @@ -59,10 +63,15 @@ import org.springframework.util.StringUtils; */ public class FileSplitter extends AbstractMessageSplitter { + private static final JsonObjectMapper objectMapper = + JsonObjectMapperProvider.jsonAvailable() ? JsonObjectMapperProvider.newInstance() : null; + private final boolean iterator; private final boolean markers; + private final boolean markersJson; + private Charset charset; /** @@ -94,11 +103,33 @@ public class FileSplitter extends AbstractMessageSplitter { * @since 4.1.5 */ public FileSplitter(boolean iterator, boolean markers) { + this(iterator, markers, false); + } + + /** + * Construct a splitter where the {@link #splitMessage(Message)} method returns an + * iterator, and the file is read line-by-line during iteration, or a list of lines + * from the file. When file markers are enabled (START/END) + * {@link #setApplySequence(boolean) applySequence} is false by default. If enabled, + * the markers are included in the sequence size. + * @param iterator true to return an iterator, false to return a list of lines. + * @param markers true to emit start of file/end of file marker messages before/after + * the data. + * @param markersJson when true, markers are represented as JSON - requires a + * supported JSON implementation on the classpath. See + * {@link JsonObjectMapperProvider} for supported implementations. + * @since 4.2.7 + */ + public FileSplitter(boolean iterator, boolean markers, boolean markersJson) { this.iterator = iterator; this.markers = markers; if (markers) { setApplySequence(false); + if (markersJson) { + Assert.notNull(objectMapper, "'markersJson' requires an object mapper"); + } } + this.markersJson = markersJson; } /** @@ -196,7 +227,9 @@ public class FileSplitter extends AbstractMessageSplitter { bufferedReader.close(); this.done = true; } - catch (IOException e1) {} + catch (IOException e1) { + // ignore + } throw new MessageHandlingException(message, "IOException while iterating", e); } } @@ -209,13 +242,13 @@ public class FileSplitter extends AbstractMessageSplitter { this.hasNextCalled = false; if (this.sof) { this.sof = false; - return new FileMarker(filePath, Mark.START, 0); + return markerToReturn(new FileMarker(filePath, Mark.START, 0)); } if (this.eof) { this.eof = false; this.markers = false; this.done = true; - return new FileMarker(filePath, Mark.END, this.lineCount); + return markerToReturn(new FileMarker(filePath, Mark.END, this.lineCount)); } if (this.line != null) { String line = this.line; @@ -229,6 +262,24 @@ public class FileSplitter extends AbstractMessageSplitter { } } + private Message markerToReturn(FileMarker fileMarker) { + Object payload; + if (FileSplitter.this.markersJson) { + try { + payload = objectMapper.toJson(fileMarker); + } + catch (Exception e) { + throw new MessageHandlingException(message, "Failed to convert marker to JSON", e); + } + } + else { + payload = fileMarker; + } + return getMessageBuilderFactory().withPayload(payload) + .setHeader(FileHeaders.MARKER, fileMarker.mark.name()) + .build(); + } + }; if (this.iterator) { @@ -294,6 +345,15 @@ public class FileSplitter extends AbstractMessageSplitter { private final long lineCount; + /* + * Provided solely to allow deserialization from JSON + */ + public FileMarker() { + this.filePath = null; + this.mark = null; + this.lineCount = 0; + } + public FileMarker(String filePath, Mark mark, long lineCount) { this.filePath = filePath; this.mark = mark; 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 3f37a133f7..b5e8274d9a 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 @@ -35,7 +35,6 @@ import java.io.Reader; import java.nio.charset.Charset; import java.util.Date; -import org.hamcrest.Matchers; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -51,6 +50,8 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.splitter.FileSplitter.FileMarker; +import org.springframework.integration.support.json.JsonObjectMapper; +import org.springframework.integration.support.json.JsonObjectMapperProvider; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -189,7 +190,8 @@ public class FileSplitterTests { Message received = outputChannel.receive(0); assertNotNull(received); assertNull(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)); - assertThat(received.getPayload(), Matchers.instanceOf(FileSplitter.FileMarker.class)); + 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()); @@ -197,13 +199,43 @@ public class FileSplitterTests { assertNotNull(outputChannel.receive(0)); received = outputChannel.receive(0); assertNotNull(received); - assertThat(received.getPayload(), Matchers.instanceOf(FileSplitter.FileMarker.class)); + assertEquals("END", received.getHeaders().get(FileHeaders.MARKER)); + 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(2, fileMarker.getLineCount()); } + @Test + public void testMarkersJson() throws Exception { + JsonObjectMapper objectMapper = JsonObjectMapperProvider.newInstance(); + QueueChannel outputChannel = new QueueChannel(); + FileSplitter splitter = new FileSplitter(true, true, true); + splitter.setOutputChannel(outputChannel); + splitter.handleMessage(new GenericMessage(file)); + Message received = outputChannel.receive(0); + assertNotNull(received); + assertNull(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)); + assertEquals("START", received.getHeaders().get(FileHeaders.MARKER)); + assertThat(received.getPayload(), instanceOf(String.class)); + String payload = (String) received.getPayload(); + assertThat(payload, containsString("\"mark\":\"START\",\"lineCount\":0")); + FileMarker fileMarker = objectMapper.fromJson(payload, FileSplitter.FileMarker.class); + assertEquals(FileSplitter.FileMarker.Mark.START, fileMarker.getMark()); + assertEquals(file.getAbsolutePath(), fileMarker.getFilePath()); + assertNotNull(outputChannel.receive(0)); + assertNotNull(outputChannel.receive(0)); + received = outputChannel.receive(0); + assertNotNull(received); + assertEquals("END", received.getHeaders().get(FileHeaders.MARKER)); + assertThat(received.getPayload(), instanceOf(String.class)); + fileMarker = objectMapper.fromJson((String) received.getPayload(), FileSplitter.FileMarker.class); + assertEquals(FileSplitter.FileMarker.Mark.END, fileMarker.getMark()); + assertEquals(file.getAbsolutePath(), fileMarker.getFilePath()); + assertEquals(2, 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 01204bbf7e..3e98736fa2 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -750,6 +750,7 @@ Other payload types will be emitted unchanged. 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. +In addition, a header `file_marker` containing `START` or `END` are added to these messages. The 'END' marker includes a line count. Default: `false`. When `true`, `apply-sequence` is `false` by default. @@ -791,3 +792,12 @@ public MessageHandler fileSplitter() { return splitter; } ---- + +When using Java configuration, an additional constructor is available: + +[source, java] +---- +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 implementation such as Jackson or Boon is on the classpath.