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
This commit is contained in:
Gary Russell
2016-06-08 13:03:09 -04:00
committed by Artem Bilan
parent 80e1a56653
commit 33c866fbb0
6 changed files with 132 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -11,6 +11,7 @@
.pmd
.project
.settings
.checkstyle
bin
build
build.log

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

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

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

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

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

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

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

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

@@ -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.