From d521031a7d370c67125403e49932fb9bdffe8c26 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 8 Jun 2016 13:03:09 -0400 Subject: [PATCH] INT-4049: FileSplitter: JSON File Markers JIRA: https://jira.spring.io/browse/INT-4049 INT-4049: Json FileMarker Namespace Support * Address PR comments --- .../json/JsonObjectMapperProvider.java | 14 ++++ .../integration/file/FileHeaders.java | 5 ++ .../file/config/FileSplitterParser.java | 1 + .../file/splitter/FileSplitter.java | 67 ++++++++++++++++++- .../config/spring-integration-file-4.3.xsd | 14 ++++ .../FileSplitterParserTests-context.xml | 1 + .../file/config/FileSplitterParserTests.java | 1 + .../file/splitter/FileSplitterTests.java | 38 ++++++++++- .../FtpServerOutboundTests-context.xml | 2 +- src/reference/asciidoc/file.adoc | 51 +++++++++----- src/reference/asciidoc/whats-new.adoc | 6 +- 11 files changed, 172 insertions(+), 28 deletions(-) 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 c5f6fab78f..7967a7c96c 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 @@ -43,6 +43,11 @@ public final class 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(); @@ -55,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 faa0e6b2f9..b561d08ce7 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 @@ -45,4 +45,9 @@ public abstract class FileHeaders { public static final String SET_MODIFIED = PREFIX + "setModified"; + /** + * 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/config/FileSplitterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java index fddad7a935..42d9bd9b3a 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 @@ -37,6 +37,7 @@ public class FileSplitterParser extends AbstractConsumerEndpointParser { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FileSplitter.class); builder.addConstructorArgValue(element.getAttribute("iterator")); builder.addConstructorArgValue(element.getAttribute("markers")); + builder.addConstructorArgValue(element.getAttribute("markers-json")); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); 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 84131da79e..5c2d7e5081 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 @@ -38,10 +38,16 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor; 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.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.MessageBuilderFactory; +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. @@ -61,10 +67,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; /** @@ -96,11 +107,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; } /** @@ -214,7 +247,9 @@ public class FileSplitter extends AbstractMessageSplitter { bufferedReader.close(); this.done = true; } - catch (IOException e1) { } + catch (IOException e1) { + // ignored + } throw new MessageHandlingException(message, "IOException while iterating", e); } } @@ -227,13 +262,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; @@ -247,6 +282,23 @@ public class FileSplitter extends AbstractMessageSplitter { } } + private AbstractIntegrationMessageBuilder 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()); + } + }; if (this.iterator) { @@ -312,6 +364,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/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd index e6bc90b707..90e862007d 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd @@ -717,8 +717,22 @@ Only files matching this regular expression will be picked up by this adapter. are filtered. The 'END' marker includes a line count. They enable the downstream processing to know when a file has been completely processed. + A header 'file_marker' is also added, containing START/END appropriately. Default: 'false'. When 'true', 'apply-sequence' is 'false' by default. + Also see 'markers-json'. + + + + + + + + + + When 'markers' is true, if this is 'true', the message payload of the marker + is a JSON String representation of the marker object. + Requires a supported JSON processor library on the classpath (Jackson, Boon). 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 44066d9884..5818b86f72 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 @@ -12,6 +12,7 @@ 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/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml index ad9f4d7180..f9ac5780a3 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml @@ -123,7 +123,7 @@ + expression="headers['file_marker']?.equals('END') ? headers['file_remoteSession'].close() : null"/> diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 85333cee66..db3070bdf8 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -799,15 +799,16 @@ Other payload types will be emitted unchanged. 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> + markers-json="" <4> + apply-sequence="" <5> + requires-reply="" <6> + charset="" <7> + input-channel="" <8> + output-channel="" <9> + send-timeout="" <10> + auto-startup="" <11> + order="" <12> + phase="" /> <13> ---- <1> The bean name of the splitter. @@ -818,33 +819,38 @@ 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. +Also see `markers-json`. -<4> Set to `false` to disable the inclusion of `sequenceSize` and `sequenceNumber` headers in messages. +<4> When `markers` is true, set this to `true` and the `FileMarker` objects will be converted to a JSON String. +Requires a supported JSON processor library on the classpath (Jackson, Boon). + +<5> 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. +<6> 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. +<7> 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 input channel used to send messages to the splitter. -<8> Set the output channel to which messages will be sent. +<9> 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 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. +<11> 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 order of this endpoint if the `input-channel` is a ``. -<12> Set the startup phase for the splitter (used when `auto-startup` is `true`). +<13> Set the startup phase for the splitter (used when `auto-startup` is `true`). *Java Configuration* @@ -866,3 +872,12 @@ using the `stream` option to retrieve a file, starting with _version 4.3_, the s the session supporting the stream, when the file is completely consumed. See <> and <> as well as <> and <> for more information about these facilities. + +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 processor library, such as Jackson or Boon, is on the classpath. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d1332fa71d..e8d900bcb3 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -149,14 +149,14 @@ See <> for more information. ===== Preserving Timestamps -The outbound channel adapter can now be configured to set the destination file's lastmodified timestamp. +The outbound channel adapter can now be configured to set the destination file's `lastmodified` timestamp. See <> for more information. ===== Splitter Changes The `FileSplitter` will now automatically close an (S)FTP session when the file is completely read. -This applies when the outbound gateway returns an `InputStream` or the new (S)FTP streaming channel adapters are being -used. +This applies when the outbound gateway returns an `InputStream` or the new (S)FTP streaming channel adapters are being used. +Also a new `markers-json` options has been introduced to convert `FileSplitter.FileMarker` to JSON `String` for relaxed downstream network interaction. See <> for more information. ==== AMQP Changes