Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4049: FileSplitter: JSON File Markers

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4049

INT-4049: Json FileMarker Namespace Support

* Address PR comments
This commit is contained in:
Gary Russell
2016-06-08 13:03:09 -04:00
committed by Artem Bilan
parent c620438913
commit d521031a7d
11 changed files with 172 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java
View File

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

5
spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java
View File

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

1
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java
View File

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

67
spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java
View File

@@ -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<Object> 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;

14
spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd
View File

@@ -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'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="markers-json" use="optional" default="false">
<xsd:annotation>
<xsd:documentation>
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).
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>

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

@@ -12,6 +12,7 @@
<int-file:splitter id="fullBoat"
iterator="false"
markers="true"
markers-json="true"
apply-sequence="true"
requires-reply="true"
charset="UTF-8"

1
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java
View File

@@ -60,6 +60,7 @@ public class FileSplitterParserTests {
public void testComplete() {
assertFalse(TestUtils.getPropertyValue(this.splitter, "iterator", Boolean.class));
assertTrue(TestUtils.getPropertyValue(this.splitter, "markers", Boolean.class));
assertTrue(TestUtils.getPropertyValue(this.splitter, "markersJson", 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"));

38
spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java
View File

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

2
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml
View File

@@ -123,7 +123,7 @@
</int:chain>
<int:service-activator input-channel="markers"
expression="payload.mark.toString().equals('END') ? headers['file_remoteSession'].close() : null"/>
expression="headers['file_marker']?.equals('END') ? headers['file_remoteSession'].close() : null"/>
<int:channel id="appending" />

51
src/reference/asciidoc/file.adoc
View File

@@ -799,15 +799,16 @@ Other payload types will be emitted unchanged.
<int-file:splitter id="splitter" <1>
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 `<publish-subscribe-channel/>`.
<12> Set the order of this endpoint if the `input-channel` is a `<publish-subscribe-channel/>`.
<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 <<ftp-streaming>> and <<sftp-streaming>> as well as <<ftp-outbound-gateway>> and <<sftp-outbound-gateway>> 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.

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -149,14 +149,14 @@ See <<file-flushing>> 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 <<file-timestamps>> 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 <<file-splitter>> for more information.
==== AMQP Changes