Add streaming json item writer
Resolves BATCH-2692
This commit is contained in:
Mahmoud Ben Hassine
committed by
Michael Minella
Michael Minella
parent
1f900ee83b
commit
84d349ea9d
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2006-2012 the original author or authors.
|
||||
* Copyright 2006-2018 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.
|
||||
@@ -16,29 +16,10 @@
|
||||
|
||||
package org.springframework.batch.item.file;
|
||||
|
||||
import java.io.BufferedWriter;
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.Writer;
|
||||
import java.nio.channels.Channels;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.charset.UnsupportedCharsetException;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
import org.springframework.batch.item.WriteFailedException;
|
||||
import org.springframework.batch.item.WriterNotOpenException;
|
||||
import org.springframework.batch.item.file.transform.LineAggregator;
|
||||
import org.springframework.batch.item.support.AbstractItemStreamItemWriter;
|
||||
import org.springframework.batch.item.util.FileUtils;
|
||||
import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.batch.item.support.AbstractFileItemWriter;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ClassUtils;
|
||||
@@ -57,48 +38,11 @@ import org.springframework.util.ClassUtils;
|
||||
* @author Robert Kasanicky
|
||||
* @author Dave Syer
|
||||
* @author Michael Minella
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class FlatFileItemWriter<T> extends AbstractItemStreamItemWriter<T> implements ResourceAwareItemWriterItemStream<T>,
|
||||
InitializingBean {
|
||||
public class FlatFileItemWriter<T> extends AbstractFileItemWriter<T> {
|
||||
|
||||
public static final boolean DEFAULT_TRANSACTIONAL = true;
|
||||
|
||||
protected static final Log logger = LogFactory.getLog(FlatFileItemWriter.class);
|
||||
|
||||
public static final String DEFAULT_LINE_SEPARATOR = System.getProperty("line.separator");
|
||||
|
||||
// default encoding for writing to output files - set to UTF-8.
|
||||
public static final String DEFAULT_CHARSET = "UTF-8";
|
||||
|
||||
private static final String WRITTEN_STATISTICS_NAME = "written";
|
||||
|
||||
private static final String RESTART_DATA_NAME = "current.count";
|
||||
|
||||
private Resource resource;
|
||||
|
||||
private OutputState state = null;
|
||||
|
||||
private LineAggregator<T> lineAggregator;
|
||||
|
||||
private boolean saveState = true;
|
||||
|
||||
private boolean forceSync = false;
|
||||
|
||||
private boolean shouldDeleteIfExists = true;
|
||||
|
||||
private boolean shouldDeleteIfEmpty = false;
|
||||
|
||||
private String encoding = DEFAULT_CHARSET;
|
||||
|
||||
private FlatFileHeaderCallback headerCallback;
|
||||
|
||||
private FlatFileFooterCallback footerCallback;
|
||||
|
||||
private String lineSeparator = DEFAULT_LINE_SEPARATOR;
|
||||
|
||||
private boolean transactional = DEFAULT_TRANSACTIONAL;
|
||||
|
||||
private boolean append = false;
|
||||
protected LineAggregator<T> lineAggregator;
|
||||
|
||||
public FlatFileItemWriter() {
|
||||
this.setExecutionContextName(ClassUtils.getShortName(FlatFileItemWriter.class));
|
||||
@@ -117,28 +61,6 @@ InitializingBean {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that changes should be force-synced to disk on flush.
|
||||
* Defaults to false, which means that even with a local disk changes could
|
||||
* be lost if the OS crashes in between a write and a cache flush. Setting
|
||||
* to true may result in slower performance for usage patterns involving many
|
||||
* frequent writes.
|
||||
*
|
||||
* @param forceSync the flag value to set
|
||||
*/
|
||||
public void setForceSync(boolean forceSync) {
|
||||
this.forceSync = forceSync;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the line separator. Defaults to the System property
|
||||
* line.separator.
|
||||
* @param lineSeparator the line separator to set
|
||||
*/
|
||||
public void setLineSeparator(String lineSeparator) {
|
||||
this.lineSeparator = lineSeparator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link LineAggregator}. This will be used to
|
||||
* translate the item into a line for output.
|
||||
@@ -149,509 +71,13 @@ InitializingBean {
|
||||
this.lineAggregator = lineAggregator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for resource. Represents a file that can be written.
|
||||
*
|
||||
* @param resource the resource to be written to
|
||||
*/
|
||||
@Override
|
||||
public void setResource(Resource resource) {
|
||||
this.resource = resource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets encoding for output template.
|
||||
*
|
||||
* @param newEncoding {@link String} containing the encoding to be used for
|
||||
* the writer.
|
||||
*/
|
||||
public void setEncoding(String newEncoding) {
|
||||
this.encoding = newEncoding;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that the target file should be deleted if it already
|
||||
* exists, otherwise it will be created. Defaults to true, so no appending
|
||||
* except on restart. If set to false and {@link #setAppendAllowed(boolean)
|
||||
* appendAllowed} is also false then there will be an exception when the
|
||||
* stream is opened to prevent existing data being potentially corrupted.
|
||||
*
|
||||
* @param shouldDeleteIfExists the flag value to set
|
||||
*/
|
||||
public void setShouldDeleteIfExists(boolean shouldDeleteIfExists) {
|
||||
this.shouldDeleteIfExists = shouldDeleteIfExists;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that the target file should be appended if it already
|
||||
* exists. If this flag is set then the flag
|
||||
* {@link #setShouldDeleteIfExists(boolean) shouldDeleteIfExists} is
|
||||
* automatically set to false, so that flag should not be set explicitly.
|
||||
* Defaults value is false.
|
||||
*
|
||||
* @param append the flag value to set
|
||||
*/
|
||||
public void setAppendAllowed(boolean append) {
|
||||
this.append = append;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that the target file should be deleted if no lines have
|
||||
* been written (other than header and footer) on close. Defaults to false.
|
||||
*
|
||||
* @param shouldDeleteIfEmpty the flag value to set
|
||||
*/
|
||||
public void setShouldDeleteIfEmpty(boolean shouldDeleteIfEmpty) {
|
||||
this.shouldDeleteIfEmpty = shouldDeleteIfEmpty;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the flag indicating whether or not state should be saved in the
|
||||
* provided {@link ExecutionContext} during the {@link ItemStream} call to
|
||||
* update. Setting this to false means that it will always start at the
|
||||
* beginning on a restart.
|
||||
*
|
||||
* @param saveState if true, state will be persisted
|
||||
*/
|
||||
public void setSaveState(boolean saveState) {
|
||||
this.saveState = saveState;
|
||||
}
|
||||
|
||||
/**
|
||||
* headerCallback will be called before writing the first item to file.
|
||||
* Newline will be automatically appended after the header is written.
|
||||
*
|
||||
* @param headerCallback {@link FlatFileHeaderCallback} to generate the header
|
||||
*
|
||||
*/
|
||||
public void setHeaderCallback(FlatFileHeaderCallback headerCallback) {
|
||||
this.headerCallback = headerCallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* footerCallback will be called after writing the last item to file, but
|
||||
* before the file is closed.
|
||||
*
|
||||
* @param footerCallback {@link FlatFileFooterCallback} to generate the footer
|
||||
*
|
||||
*/
|
||||
public void setFooterCallback(FlatFileFooterCallback footerCallback) {
|
||||
this.footerCallback = footerCallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that writing to the buffer should be delayed if a
|
||||
* transaction is active. Defaults to true.
|
||||
*
|
||||
* @param transactional true if writing to buffer should be delayed.
|
||||
*
|
||||
*/
|
||||
public void setTransactional(boolean transactional) {
|
||||
this.transactional = transactional;
|
||||
}
|
||||
|
||||
/**
|
||||
* Writes out a string followed by a "new line", where the format of the new
|
||||
* line separator is determined by the underlying operating system. If the
|
||||
* input is not a String and a converter is available the converter will be
|
||||
* applied and then this method recursively called with the result. If the
|
||||
* input is an array or collection each value will be written to a separate
|
||||
* line (recursively calling this method for each value). If no converter is
|
||||
* supplied the input object's toString method will be used.<br>
|
||||
*
|
||||
* @param items list of items to be written to output stream
|
||||
* @throws Exception if the transformer or file output fail,
|
||||
* WriterNotOpenException if the writer has not been initialized.
|
||||
*/
|
||||
@Override
|
||||
public void write(List<? extends T> items) throws Exception {
|
||||
|
||||
if (!getOutputState().isInitialized()) {
|
||||
throw new WriterNotOpenException("Writer must be open before it can be written to");
|
||||
}
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Writing to flat file with " + items.size() + " items.");
|
||||
}
|
||||
|
||||
OutputState state = getOutputState();
|
||||
|
||||
public String doWrite(List<? extends T> items) {
|
||||
StringBuilder lines = new StringBuilder();
|
||||
int lineCount = 0;
|
||||
for (T item : items) {
|
||||
lines.append(lineAggregator.aggregate(item) + lineSeparator);
|
||||
lineCount++;
|
||||
lines.append(this.lineAggregator.aggregate(item)).append(this.lineSeparator);
|
||||
}
|
||||
try {
|
||||
state.write(lines.toString());
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new WriteFailedException("Could not write data. The file may be corrupt.", e);
|
||||
}
|
||||
state.linesWritten += lineCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see ItemStream#close()
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
super.close();
|
||||
if (state != null) {
|
||||
try {
|
||||
if (footerCallback != null && state.outputBufferedWriter != null) {
|
||||
footerCallback.writeFooter(state.outputBufferedWriter);
|
||||
state.outputBufferedWriter.flush();
|
||||
}
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Failed to write footer before closing", e);
|
||||
}
|
||||
finally {
|
||||
state.close();
|
||||
if (state.linesWritten == 0 && shouldDeleteIfEmpty) {
|
||||
try {
|
||||
resource.getFile().delete();
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Failed to delete empty file on close", e);
|
||||
}
|
||||
}
|
||||
state = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the reader. This method may be called multiple times before
|
||||
* close is called.
|
||||
*
|
||||
* @see ItemStream#open(ExecutionContext)
|
||||
*/
|
||||
@Override
|
||||
public void open(ExecutionContext executionContext) throws ItemStreamException {
|
||||
super.open(executionContext);
|
||||
|
||||
Assert.notNull(resource, "The resource must be set");
|
||||
|
||||
if (!getOutputState().isInitialized()) {
|
||||
doOpen(executionContext);
|
||||
}
|
||||
}
|
||||
|
||||
private void doOpen(ExecutionContext executionContext) throws ItemStreamException {
|
||||
OutputState outputState = getOutputState();
|
||||
if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) {
|
||||
outputState.restoreFrom(executionContext);
|
||||
}
|
||||
try {
|
||||
outputState.initializeBufferedWriter();
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Failed to initialize writer", ioe);
|
||||
}
|
||||
if (outputState.lastMarkedByteOffsetPosition == 0 && !outputState.appending) {
|
||||
if (headerCallback != null) {
|
||||
try {
|
||||
headerCallback.writeHeader(outputState.outputBufferedWriter);
|
||||
outputState.write(lineSeparator);
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Could not write headers. The file may be corrupt.", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @see ItemStream#update(ExecutionContext)
|
||||
*/
|
||||
@Override
|
||||
public void update(ExecutionContext executionContext) {
|
||||
super.update(executionContext);
|
||||
if (state == null) {
|
||||
throw new ItemStreamException("ItemStream not open or already closed.");
|
||||
}
|
||||
|
||||
Assert.notNull(executionContext, "ExecutionContext must not be null");
|
||||
|
||||
if (saveState) {
|
||||
|
||||
try {
|
||||
executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), state.position());
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("ItemStream does not return current position properly", e);
|
||||
}
|
||||
|
||||
executionContext.putLong(getExecutionContextKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
|
||||
}
|
||||
}
|
||||
|
||||
// Returns object representing state.
|
||||
private OutputState getOutputState() {
|
||||
if (state == null) {
|
||||
File file;
|
||||
try {
|
||||
file = resource.getFile();
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Could not convert resource to file: [" + resource + "]", e);
|
||||
}
|
||||
Assert.state(!file.exists() || file.canWrite(), "Resource is not writable: [" + resource + "]");
|
||||
state = new OutputState();
|
||||
state.setDeleteIfExists(shouldDeleteIfExists);
|
||||
state.setAppendAllowed(append);
|
||||
state.setEncoding(encoding);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encapsulates the runtime state of the writer. All state changing
|
||||
* operations on the writer go through this class.
|
||||
*/
|
||||
private class OutputState {
|
||||
|
||||
private FileOutputStream os;
|
||||
|
||||
// The bufferedWriter over the file channel that is actually written
|
||||
Writer outputBufferedWriter;
|
||||
|
||||
FileChannel fileChannel;
|
||||
|
||||
// this represents the charset encoding (if any is needed) for the
|
||||
// output file
|
||||
String encoding = DEFAULT_CHARSET;
|
||||
|
||||
boolean restarted = false;
|
||||
|
||||
long lastMarkedByteOffsetPosition = 0;
|
||||
|
||||
long linesWritten = 0;
|
||||
|
||||
boolean shouldDeleteIfExists = true;
|
||||
|
||||
boolean initialized = false;
|
||||
|
||||
private boolean append = false;
|
||||
|
||||
private boolean appending = false;
|
||||
|
||||
/**
|
||||
* Return the byte offset position of the cursor in the output file as a
|
||||
* long integer.
|
||||
*/
|
||||
public long position() throws IOException {
|
||||
long pos = 0;
|
||||
|
||||
if (fileChannel == null) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
outputBufferedWriter.flush();
|
||||
pos = fileChannel.position();
|
||||
if (transactional) {
|
||||
pos += ((TransactionAwareBufferedWriter) outputBufferedWriter).getBufferSize();
|
||||
}
|
||||
|
||||
return pos;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param append if true, append to previously created file
|
||||
*/
|
||||
public void setAppendAllowed(boolean append) {
|
||||
this.append = append;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param executionContext state from which to restore writing from
|
||||
*/
|
||||
public void restoreFrom(ExecutionContext executionContext) {
|
||||
lastMarkedByteOffsetPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME));
|
||||
linesWritten = executionContext.getLong(getExecutionContextKey(WRITTEN_STATISTICS_NAME));
|
||||
if (shouldDeleteIfEmpty && linesWritten == 0) {
|
||||
// previous execution deleted the output file because no items were written
|
||||
restarted = false;
|
||||
lastMarkedByteOffsetPosition = 0;
|
||||
} else {
|
||||
restarted = true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param shouldDeleteIfExists indicator
|
||||
*/
|
||||
public void setDeleteIfExists(boolean shouldDeleteIfExists) {
|
||||
this.shouldDeleteIfExists = shouldDeleteIfExists;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param encoding file encoding
|
||||
*/
|
||||
public void setEncoding(String encoding) {
|
||||
this.encoding = encoding;
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the open resource and reset counters.
|
||||
*/
|
||||
public void close() {
|
||||
|
||||
initialized = false;
|
||||
restarted = false;
|
||||
try {
|
||||
if (outputBufferedWriter != null) {
|
||||
outputBufferedWriter.close();
|
||||
}
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
|
||||
}
|
||||
finally {
|
||||
if (!transactional) {
|
||||
closeStream();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void closeStream() {
|
||||
try {
|
||||
if (fileChannel != null) {
|
||||
fileChannel.close();
|
||||
}
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
|
||||
}
|
||||
finally {
|
||||
try {
|
||||
if (os != null) {
|
||||
os.close();
|
||||
}
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param line String to be written to the file
|
||||
* @throws IOException
|
||||
*/
|
||||
public void write(String line) throws IOException {
|
||||
if (!initialized) {
|
||||
initializeBufferedWriter();
|
||||
}
|
||||
|
||||
outputBufferedWriter.write(line);
|
||||
outputBufferedWriter.flush();
|
||||
}
|
||||
|
||||
/**
|
||||
* Truncate the output at the last known good point.
|
||||
*
|
||||
* @throws IOException if unable to work with file
|
||||
*/
|
||||
public void truncate() throws IOException {
|
||||
fileChannel.truncate(lastMarkedByteOffsetPosition);
|
||||
fileChannel.position(lastMarkedByteOffsetPosition);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the buffered writer for the output file channel based on
|
||||
* configuration information.
|
||||
* @throws IOException if unable to initialize buffer
|
||||
*/
|
||||
private void initializeBufferedWriter() throws IOException {
|
||||
|
||||
File file = resource.getFile();
|
||||
FileUtils.setUpOutputFile(file, restarted, append, shouldDeleteIfExists);
|
||||
|
||||
os = new FileOutputStream(file.getAbsolutePath(), true);
|
||||
fileChannel = os.getChannel();
|
||||
|
||||
outputBufferedWriter = getBufferedWriter(fileChannel, encoding);
|
||||
outputBufferedWriter.flush();
|
||||
|
||||
if (append) {
|
||||
// Bug in IO library? This doesn't work...
|
||||
// lastMarkedByteOffsetPosition = fileChannel.position();
|
||||
if (file.length() > 0) {
|
||||
appending = true;
|
||||
// Don't write the headers again
|
||||
}
|
||||
}
|
||||
|
||||
Assert.state(outputBufferedWriter != null,
|
||||
"Unable to initialize buffered writer");
|
||||
// in case of restarting reset position to last committed point
|
||||
if (restarted) {
|
||||
checkFileSize();
|
||||
truncate();
|
||||
}
|
||||
|
||||
initialized = true;
|
||||
}
|
||||
|
||||
public boolean isInitialized() {
|
||||
return initialized;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the buffered writer opened to the beginning of the file
|
||||
* specified by the absolute path name contained in absoluteFileName.
|
||||
*/
|
||||
private Writer getBufferedWriter(FileChannel fileChannel, String encoding) {
|
||||
try {
|
||||
final FileChannel channel = fileChannel;
|
||||
if (transactional) {
|
||||
TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, () -> closeStream());
|
||||
|
||||
writer.setEncoding(encoding);
|
||||
writer.setForceSync(forceSync);
|
||||
return writer;
|
||||
}
|
||||
else {
|
||||
Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) {
|
||||
@Override
|
||||
public void flush() throws IOException {
|
||||
super.flush();
|
||||
if (forceSync) {
|
||||
channel.force(false);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
return writer;
|
||||
}
|
||||
}
|
||||
catch (UnsupportedCharsetException ucse) {
|
||||
throw new ItemStreamException("Bad encoding configuration for output file " + fileChannel, ucse);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks (on setState) to make sure that the current output file's size
|
||||
* is not smaller than the last saved commit point. If it is, then the
|
||||
* file has been damaged in some way and whole task must be started over
|
||||
* again from the beginning.
|
||||
* @throws IOException if there is an IO problem
|
||||
*/
|
||||
private void checkFileSize() throws IOException {
|
||||
long size = -1;
|
||||
|
||||
outputBufferedWriter.flush();
|
||||
size = fileChannel.size();
|
||||
|
||||
if (size < lastMarkedByteOffsetPosition) {
|
||||
throw new ItemStreamException("Current file size is smaller than size at last commit");
|
||||
}
|
||||
}
|
||||
|
||||
return lines.toString();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
/*
|
||||
* Copyright 2018 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.json;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
|
||||
/**
|
||||
* A json object marshaller that uses <a href="https://github.com/google/gson">Google Gson</a>
|
||||
* to marshal an object into a json representation.
|
||||
*
|
||||
* @param <T> type of objects to marshal
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 4.1
|
||||
*/
|
||||
public class GsonJsonObjectMarshaller<T> implements JsonObjectMarshaller<T> {
|
||||
|
||||
private Gson gson = new Gson();
|
||||
|
||||
/**
|
||||
* Set the {@link Gson} object to use.
|
||||
* @param gson object to use
|
||||
*/
|
||||
public void setGson(Gson gson) {
|
||||
this.gson = gson;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String marshal(T item) {
|
||||
return gson.toJson(item);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
/*
|
||||
* Copyright 2018 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.json;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
|
||||
/**
|
||||
* A json object marshaller that uses <a href="https://github.com/FasterXML/jackson">Jackson</a>
|
||||
* to marshal an object into a json representation.
|
||||
*
|
||||
* @param <T> type of objects to marshal
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 4.1
|
||||
*/
|
||||
public class JacksonJsonObjectMarshaller<T> implements JsonObjectMarshaller<T> {
|
||||
|
||||
private ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
/**
|
||||
* Set the {@link ObjectMapper} to use.
|
||||
* @param objectMapper to use
|
||||
*/
|
||||
public void setObjectMapper(ObjectMapper objectMapper) {
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String marshal(T item) {
|
||||
try {
|
||||
return objectMapper.writeValueAsString(item);
|
||||
} catch (JsonProcessingException e) {
|
||||
throw new ItemStreamException("Unable to marshal object " + item + " to Json", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
/*
|
||||
* Copyright 2018 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.json;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.batch.item.support.AbstractFileItemWriter;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
/**
|
||||
* Item writer that writes data in json format to an output file. The location
|
||||
* of the output file is defined by a {@link Resource} and must represent a
|
||||
* writable file. Items are transformed to json format using a
|
||||
* {@link JsonObjectMarshaller}. Items will be enclosed in a json array as follows:
|
||||
*
|
||||
* <p>
|
||||
* <code>
|
||||
* [
|
||||
* {json object},
|
||||
* {json object},
|
||||
* {json object}
|
||||
* ]
|
||||
* </code>
|
||||
* </p>
|
||||
*
|
||||
* The implementation is <b>not</b> thread-safe.
|
||||
*
|
||||
* @see GsonJsonObjectMarshaller
|
||||
* @see JacksonJsonObjectMarshaller
|
||||
* @param <T> type of object to write as json representation
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 4.1
|
||||
*/
|
||||
public class JsonFileItemWriter<T> extends AbstractFileItemWriter<T> {
|
||||
|
||||
private static final char JSON_OBJECT_SEPARATOR = ',';
|
||||
private static final char JSON_ARRAY_START = '[';
|
||||
private static final char JSON_ARRAY_STOP = ']';
|
||||
|
||||
private JsonObjectMarshaller<T> jsonObjectMarshaller;
|
||||
|
||||
/**
|
||||
* Create a new {@link JsonFileItemWriter} instance.
|
||||
* @param resource to write json data to
|
||||
* @param jsonObjectMarshaller used to marshal object into json representation
|
||||
*/
|
||||
public JsonFileItemWriter(Resource resource, JsonObjectMarshaller<T> jsonObjectMarshaller) {
|
||||
Assert.notNull(resource, "resource must not be null");
|
||||
Assert.notNull(jsonObjectMarshaller, "json object marshaller must not be null");
|
||||
setResource(resource);
|
||||
setJsonObjectMarshaller(jsonObjectMarshaller);
|
||||
setHeaderCallback(writer -> writer.write(JSON_ARRAY_START));
|
||||
setFooterCallback(writer -> writer.write(this.lineSeparator + JSON_ARRAY_STOP + this.lineSeparator));
|
||||
setExecutionContextName(ClassUtils.getShortName(JsonFileItemWriter.class));
|
||||
}
|
||||
|
||||
/**
|
||||
* Assert that mandatory properties (jsonObjectMarshaller) are set.
|
||||
*
|
||||
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
if (this.append) {
|
||||
this.shouldDeleteIfExists = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the {@link JsonObjectMarshaller} to use to marshal object to json.
|
||||
* @param jsonObjectMarshaller the marshaller to use
|
||||
*/
|
||||
public void setJsonObjectMarshaller(JsonObjectMarshaller<T> jsonObjectMarshaller) {
|
||||
this.jsonObjectMarshaller = jsonObjectMarshaller;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String doWrite(List<? extends T> items) {
|
||||
StringBuilder lines = new StringBuilder();
|
||||
Iterator<? extends T> iterator = items.iterator();
|
||||
while (iterator.hasNext()) {
|
||||
if (iterator.hasNext() && state.getLinesWritten() > 0) {
|
||||
lines.append(JSON_OBJECT_SEPARATOR).append(this.lineSeparator);
|
||||
}
|
||||
T item = iterator.next();
|
||||
lines.append(' ').append(this.jsonObjectMarshaller.marshal(item));
|
||||
if (iterator.hasNext()) {
|
||||
lines.append(JSON_OBJECT_SEPARATOR).append(this.lineSeparator);
|
||||
}
|
||||
}
|
||||
return lines.toString();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
/*
|
||||
* Copyright 2018 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.json;
|
||||
|
||||
/**
|
||||
* Strategy interface to marshal an object into a json representation.
|
||||
* Implementations are required to return a valid json object.
|
||||
*
|
||||
* @param <T> type of objects to marshal
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 4.1
|
||||
*/
|
||||
public interface JsonObjectMarshaller<T> {
|
||||
|
||||
/**
|
||||
* Marshal an object into a json representation.
|
||||
* @param object to marshal
|
||||
* @return json representation fo the object
|
||||
*/
|
||||
String marshal(T object);
|
||||
|
||||
}
|
||||
@@ -0,0 +1,261 @@
|
||||
/*
|
||||
* Copyright 2018 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.json.builder;
|
||||
|
||||
import org.springframework.batch.item.file.FlatFileFooterCallback;
|
||||
import org.springframework.batch.item.file.FlatFileHeaderCallback;
|
||||
import org.springframework.batch.item.json.JsonFileItemWriter;
|
||||
import org.springframework.batch.item.json.JsonObjectMarshaller;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Builder for {@link JsonFileItemWriter}.
|
||||
*
|
||||
* @param <T> type of objects to write as Json output.
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 4.1
|
||||
*/
|
||||
public class JsonFileItemWriterBuilder<T> {
|
||||
|
||||
private Resource resource;
|
||||
private JsonObjectMarshaller<T> jsonObjectMarshaller;
|
||||
private FlatFileHeaderCallback headerCallback;
|
||||
private FlatFileFooterCallback footerCallback;
|
||||
|
||||
private String name;
|
||||
private String encoding = JsonFileItemWriter.DEFAULT_CHARSET;
|
||||
private String lineSeparator = JsonFileItemWriter.DEFAULT_LINE_SEPARATOR;
|
||||
|
||||
private boolean append = false;
|
||||
private boolean forceSync = false;
|
||||
private boolean saveState = true;
|
||||
private boolean shouldDeleteIfExists = true;
|
||||
private boolean shouldDeleteIfEmpty = false;
|
||||
private boolean transactional = JsonFileItemWriter.DEFAULT_TRANSACTIONAL;
|
||||
|
||||
/**
|
||||
* Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport}
|
||||
* should be persisted within the {@link org.springframework.batch.item.ExecutionContext}
|
||||
* for restart purposes.
|
||||
*
|
||||
* @param saveState defaults to true
|
||||
* @return The current instance of the builder.
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> saveState(boolean saveState) {
|
||||
this.saveState = saveState;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The name used to calculate the key within the
|
||||
* {@link org.springframework.batch.item.ExecutionContext}. Required if
|
||||
* {@link #saveState(boolean)} is set to true.
|
||||
*
|
||||
* @param name name of the reader instance
|
||||
* @return The current instance of the builder.
|
||||
* @see org.springframework.batch.item.ItemStreamSupport#setName(String)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> name(String name) {
|
||||
this.name = name;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* A flag indicating that changes should be force-synced to disk on flush. Defaults
|
||||
* to false.
|
||||
*
|
||||
* @param forceSync value to set the flag to
|
||||
* @return The current instance of the builder.
|
||||
* @see JsonFileItemWriter#setForceSync(boolean)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> forceSync(boolean forceSync) {
|
||||
this.forceSync = forceSync;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* String used to separate lines in output. Defaults to the System property
|
||||
* <code>line.separator</code>.
|
||||
*
|
||||
* @param lineSeparator value to use for a line separator
|
||||
* @return The current instance of the builder.
|
||||
* @see JsonFileItemWriter#setLineSeparator(String)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> lineSeparator(String lineSeparator) {
|
||||
this.lineSeparator = lineSeparator;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the {@link JsonObjectMarshaller} to use to marshal objects to json.
|
||||
*
|
||||
* @param jsonObjectMarshaller to use
|
||||
* @return The current instance of the builder.
|
||||
* @see JsonFileItemWriter#setJsonObjectMarshaller(JsonObjectMarshaller)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> jsonObjectMarshaller(JsonObjectMarshaller<T> jsonObjectMarshaller) {
|
||||
this.jsonObjectMarshaller = jsonObjectMarshaller;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The {@link Resource} to be used as output.
|
||||
*
|
||||
* @param resource the output of the writer.
|
||||
* @return The current instance of the builder.
|
||||
* @see JsonFileItemWriter#setResource(Resource)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> resource(Resource resource) {
|
||||
this.resource = resource;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encoding used for output.
|
||||
*
|
||||
* @param encoding encoding type.
|
||||
* @return The current instance of the builder.
|
||||
* @see JsonFileItemWriter#setEncoding(String)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> encoding(String encoding) {
|
||||
this.encoding = encoding;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* If set to true, once the step is complete, if the resource previously provided is
|
||||
* empty, it will be deleted.
|
||||
*
|
||||
* @param shouldDelete defaults to false
|
||||
* @return The current instance of the builder
|
||||
* @see JsonFileItemWriter#setShouldDeleteIfEmpty(boolean)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> shouldDeleteIfEmpty(boolean shouldDelete) {
|
||||
this.shouldDeleteIfEmpty = shouldDelete;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* If set to true, upon the start of the step, if the resource already exists, it will
|
||||
* be deleted and recreated.
|
||||
*
|
||||
* @param shouldDelete defaults to true
|
||||
* @return The current instance of the builder
|
||||
* @see JsonFileItemWriter#setShouldDeleteIfExists(boolean)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> shouldDeleteIfExists(boolean shouldDelete) {
|
||||
this.shouldDeleteIfExists = shouldDelete;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* If set to true and the file exists, the output will be appended to the existing
|
||||
* file.
|
||||
*
|
||||
* @param append defaults to false
|
||||
* @return The current instance of the builder
|
||||
* @see JsonFileItemWriter#setAppendAllowed(boolean)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> append(boolean append) {
|
||||
this.append = append;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* A callback for header processing.
|
||||
*
|
||||
* @param callback {@link FlatFileHeaderCallback} implementation
|
||||
* @return The current instance of the builder
|
||||
* @see JsonFileItemWriter#setHeaderCallback(FlatFileHeaderCallback)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> headerCallback(FlatFileHeaderCallback callback) {
|
||||
this.headerCallback = callback;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* A callback for footer processing.
|
||||
*
|
||||
* @param callback {@link FlatFileFooterCallback} implementation
|
||||
* @return The current instance of the builder
|
||||
* @see JsonFileItemWriter#setFooterCallback(FlatFileFooterCallback)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> footerCallback(FlatFileFooterCallback callback) {
|
||||
this.footerCallback = callback;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* If set to true, the flushing of the buffer is delayed while a transaction is active.
|
||||
*
|
||||
* @param transactional defaults to true
|
||||
* @return The current instance of the builder
|
||||
* @see JsonFileItemWriter#setTransactional(boolean)
|
||||
*/
|
||||
public JsonFileItemWriterBuilder<T> transactional(boolean transactional) {
|
||||
this.transactional = transactional;
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate the configuration and build a new {@link JsonFileItemWriter}.
|
||||
*
|
||||
* @return a new instance of the {@link JsonFileItemWriter}
|
||||
*/
|
||||
public JsonFileItemWriter<T> build() {
|
||||
Assert.notNull(this.resource, "A resource is required.");
|
||||
Assert.notNull(this.jsonObjectMarshaller, "A json object marshaller is required.");
|
||||
|
||||
if (this.saveState) {
|
||||
Assert.hasText(this.name, "A name is required when saveState is true");
|
||||
}
|
||||
|
||||
JsonFileItemWriter<T> jsonFileItemWriter = new JsonFileItemWriter<>(this.resource, this.jsonObjectMarshaller);
|
||||
|
||||
jsonFileItemWriter.setName(this.name);
|
||||
jsonFileItemWriter.setAppendAllowed(this.append);
|
||||
jsonFileItemWriter.setEncoding(this.encoding);
|
||||
if (this.headerCallback != null) {
|
||||
jsonFileItemWriter.setHeaderCallback(this.headerCallback);
|
||||
}
|
||||
if (this.footerCallback != null) {
|
||||
jsonFileItemWriter.setFooterCallback(this.footerCallback);
|
||||
}
|
||||
jsonFileItemWriter.setForceSync(this.forceSync);
|
||||
jsonFileItemWriter.setLineSeparator(this.lineSeparator);
|
||||
jsonFileItemWriter.setSaveState(this.saveState);
|
||||
jsonFileItemWriter.setShouldDeleteIfEmpty(this.shouldDeleteIfEmpty);
|
||||
jsonFileItemWriter.setShouldDeleteIfExists(this.shouldDeleteIfExists);
|
||||
jsonFileItemWriter.setTransactional(this.transactional);
|
||||
|
||||
return jsonFileItemWriter;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,636 @@
|
||||
/*
|
||||
* Copyright 2006-2018 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.item.support;
|
||||
|
||||
import java.io.BufferedWriter;
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.Writer;
|
||||
import java.nio.channels.Channels;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.charset.UnsupportedCharsetException;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
import org.springframework.batch.item.WriteFailedException;
|
||||
import org.springframework.batch.item.WriterNotOpenException;
|
||||
import org.springframework.batch.item.file.FlatFileFooterCallback;
|
||||
import org.springframework.batch.item.file.FlatFileHeaderCallback;
|
||||
import org.springframework.batch.item.file.ResourceAwareItemWriterItemStream;
|
||||
import org.springframework.batch.item.util.FileUtils;
|
||||
import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Base class for item writers that write data to a file or stream.
|
||||
* This class provides common features like restart, force sync, append etc.
|
||||
* The location of the output file is defined by a {@link Resource} which must
|
||||
* represent a writable file.<br>
|
||||
*
|
||||
* Uses buffered writer to improve performance.<br>
|
||||
*
|
||||
* The implementation is <b>not</b> thread-safe.
|
||||
*
|
||||
* @author Waseem Malik
|
||||
* @author Tomas Slanina
|
||||
* @author Robert Kasanicky
|
||||
* @author Dave Syer
|
||||
* @author Michael Minella
|
||||
* @author Mahmoud Ben Hassine
|
||||
*
|
||||
* @since 4.1
|
||||
*/
|
||||
public abstract class AbstractFileItemWriter<T> extends AbstractItemStreamItemWriter<T>
|
||||
implements ResourceAwareItemWriterItemStream<T>, InitializingBean {
|
||||
|
||||
public static final boolean DEFAULT_TRANSACTIONAL = true;
|
||||
|
||||
protected static final Log logger = LogFactory.getLog(AbstractFileItemWriter.class);
|
||||
|
||||
public static final String DEFAULT_LINE_SEPARATOR = System.getProperty("line.separator");
|
||||
|
||||
// default encoding for writing to output files - set to UTF-8.
|
||||
public static final String DEFAULT_CHARSET = "UTF-8";
|
||||
|
||||
private static final String WRITTEN_STATISTICS_NAME = "written";
|
||||
|
||||
private static final String RESTART_DATA_NAME = "current.count";
|
||||
|
||||
private Resource resource;
|
||||
|
||||
protected OutputState state = null;
|
||||
|
||||
private boolean saveState = true;
|
||||
|
||||
private boolean forceSync = false;
|
||||
|
||||
protected boolean shouldDeleteIfExists = true;
|
||||
|
||||
private boolean shouldDeleteIfEmpty = false;
|
||||
|
||||
private String encoding = DEFAULT_CHARSET;
|
||||
|
||||
private FlatFileHeaderCallback headerCallback;
|
||||
|
||||
private FlatFileFooterCallback footerCallback;
|
||||
|
||||
protected String lineSeparator = DEFAULT_LINE_SEPARATOR;
|
||||
|
||||
private boolean transactional = DEFAULT_TRANSACTIONAL;
|
||||
|
||||
protected boolean append = false;
|
||||
|
||||
/**
|
||||
* Flag to indicate that changes should be force-synced to disk on flush.
|
||||
* Defaults to false, which means that even with a local disk changes could
|
||||
* be lost if the OS crashes in between a write and a cache flush. Setting
|
||||
* to true may result in slower performance for usage patterns involving many
|
||||
* frequent writes.
|
||||
*
|
||||
* @param forceSync the flag value to set
|
||||
*/
|
||||
public void setForceSync(boolean forceSync) {
|
||||
this.forceSync = forceSync;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the line separator. Defaults to the System property
|
||||
* line.separator.
|
||||
* @param lineSeparator the line separator to set
|
||||
*/
|
||||
public void setLineSeparator(String lineSeparator) {
|
||||
this.lineSeparator = lineSeparator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for resource. Represents a file that can be written.
|
||||
*
|
||||
* @param resource the resource to be written to
|
||||
*/
|
||||
@Override
|
||||
public void setResource(Resource resource) {
|
||||
this.resource = resource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets encoding for output template.
|
||||
*
|
||||
* @param newEncoding {@link String} containing the encoding to be used for
|
||||
* the writer.
|
||||
*/
|
||||
public void setEncoding(String newEncoding) {
|
||||
this.encoding = newEncoding;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that the target file should be deleted if it already
|
||||
* exists, otherwise it will be created. Defaults to true, so no appending
|
||||
* except on restart. If set to false and {@link #setAppendAllowed(boolean)
|
||||
* appendAllowed} is also false then there will be an exception when the
|
||||
* stream is opened to prevent existing data being potentially corrupted.
|
||||
*
|
||||
* @param shouldDeleteIfExists the flag value to set
|
||||
*/
|
||||
public void setShouldDeleteIfExists(boolean shouldDeleteIfExists) {
|
||||
this.shouldDeleteIfExists = shouldDeleteIfExists;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that the target file should be appended if it already
|
||||
* exists. If this flag is set then the flag
|
||||
* {@link #setShouldDeleteIfExists(boolean) shouldDeleteIfExists} is
|
||||
* automatically set to false, so that flag should not be set explicitly.
|
||||
* Defaults value is false.
|
||||
*
|
||||
* @param append the flag value to set
|
||||
*/
|
||||
public void setAppendAllowed(boolean append) {
|
||||
this.append = append;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that the target file should be deleted if no lines have
|
||||
* been written (other than header and footer) on close. Defaults to false.
|
||||
*
|
||||
* @param shouldDeleteIfEmpty the flag value to set
|
||||
*/
|
||||
public void setShouldDeleteIfEmpty(boolean shouldDeleteIfEmpty) {
|
||||
this.shouldDeleteIfEmpty = shouldDeleteIfEmpty;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the flag indicating whether or not state should be saved in the
|
||||
* provided {@link ExecutionContext} during the {@link ItemStream} call to
|
||||
* update. Setting this to false means that it will always start at the
|
||||
* beginning on a restart.
|
||||
*
|
||||
* @param saveState if true, state will be persisted
|
||||
*/
|
||||
public void setSaveState(boolean saveState) {
|
||||
this.saveState = saveState;
|
||||
}
|
||||
|
||||
/**
|
||||
* headerCallback will be called before writing the first item to file.
|
||||
* Newline will be automatically appended after the header is written.
|
||||
*
|
||||
* @param headerCallback {@link FlatFileHeaderCallback} to generate the header
|
||||
*
|
||||
*/
|
||||
public void setHeaderCallback(FlatFileHeaderCallback headerCallback) {
|
||||
this.headerCallback = headerCallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* footerCallback will be called after writing the last item to file, but
|
||||
* before the file is closed.
|
||||
*
|
||||
* @param footerCallback {@link FlatFileFooterCallback} to generate the footer
|
||||
*
|
||||
*/
|
||||
public void setFooterCallback(FlatFileFooterCallback footerCallback) {
|
||||
this.footerCallback = footerCallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flag to indicate that writing to the buffer should be delayed if a
|
||||
* transaction is active. Defaults to true.
|
||||
*
|
||||
* @param transactional true if writing to buffer should be delayed.
|
||||
*
|
||||
*/
|
||||
public void setTransactional(boolean transactional) {
|
||||
this.transactional = transactional;
|
||||
}
|
||||
|
||||
/**
|
||||
* Writes out a string followed by a "new line", where the format of the new
|
||||
* line separator is determined by the underlying operating system.
|
||||
*
|
||||
* @param items list of items to be written to output stream
|
||||
* @throws Exception if an error occurs while writing items to the output stream
|
||||
*/
|
||||
@Override
|
||||
public void write(List<? extends T> items) throws Exception {
|
||||
if (!getOutputState().isInitialized()) {
|
||||
throw new WriterNotOpenException("Writer must be open before it can be written to");
|
||||
}
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Writing to file with " + items.size() + " items.");
|
||||
}
|
||||
|
||||
OutputState state = getOutputState();
|
||||
|
||||
String lines = doWrite(items);
|
||||
try {
|
||||
state.write(lines);
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new WriteFailedException("Could not write data. The file may be corrupt.", e);
|
||||
}
|
||||
state.setLinesWritten(state.getLinesWritten() + items.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Write out a string of items followed by a "new line", where the format of the new
|
||||
* line separator is determined by the underlying operating system.
|
||||
* @param items to be written
|
||||
* @return written lines
|
||||
*/
|
||||
protected abstract String doWrite(List<? extends T> items);
|
||||
|
||||
/**
|
||||
* @see ItemStream#close()
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
super.close();
|
||||
if (state != null) {
|
||||
try {
|
||||
if (footerCallback != null && state.outputBufferedWriter != null) {
|
||||
footerCallback.writeFooter(state.outputBufferedWriter);
|
||||
state.outputBufferedWriter.flush();
|
||||
}
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Failed to write footer before closing", e);
|
||||
}
|
||||
finally {
|
||||
state.close();
|
||||
if (state.linesWritten == 0 && shouldDeleteIfEmpty) {
|
||||
try {
|
||||
resource.getFile().delete();
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Failed to delete empty file on close", e);
|
||||
}
|
||||
}
|
||||
state = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the reader. This method may be called multiple times before
|
||||
* close is called.
|
||||
*
|
||||
* @see ItemStream#open(ExecutionContext)
|
||||
*/
|
||||
@Override
|
||||
public void open(ExecutionContext executionContext) throws ItemStreamException {
|
||||
super.open(executionContext);
|
||||
|
||||
Assert.notNull(resource, "The resource must be set");
|
||||
|
||||
if (!getOutputState().isInitialized()) {
|
||||
doOpen(executionContext);
|
||||
}
|
||||
}
|
||||
|
||||
private void doOpen(ExecutionContext executionContext) throws ItemStreamException {
|
||||
OutputState outputState = getOutputState();
|
||||
if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) {
|
||||
outputState.restoreFrom(executionContext);
|
||||
}
|
||||
try {
|
||||
outputState.initializeBufferedWriter();
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Failed to initialize writer", ioe);
|
||||
}
|
||||
if (outputState.lastMarkedByteOffsetPosition == 0 && !outputState.appending) {
|
||||
if (headerCallback != null) {
|
||||
try {
|
||||
headerCallback.writeHeader(outputState.outputBufferedWriter);
|
||||
outputState.write(lineSeparator);
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Could not write headers. The file may be corrupt.", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @see ItemStream#update(ExecutionContext)
|
||||
*/
|
||||
@Override
|
||||
public void update(ExecutionContext executionContext) {
|
||||
super.update(executionContext);
|
||||
if (state == null) {
|
||||
throw new ItemStreamException("ItemStream not open or already closed.");
|
||||
}
|
||||
|
||||
Assert.notNull(executionContext, "ExecutionContext must not be null");
|
||||
|
||||
if (saveState) {
|
||||
|
||||
try {
|
||||
executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), state.position());
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("ItemStream does not return current position properly", e);
|
||||
}
|
||||
|
||||
executionContext.putLong(getExecutionContextKey(WRITTEN_STATISTICS_NAME), state.linesWritten);
|
||||
}
|
||||
}
|
||||
|
||||
// Returns object representing state.
|
||||
protected OutputState getOutputState() {
|
||||
if (state == null) {
|
||||
File file;
|
||||
try {
|
||||
file = resource.getFile();
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new ItemStreamException("Could not convert resource to file: [" + resource + "]", e);
|
||||
}
|
||||
Assert.state(!file.exists() || file.canWrite(), "Resource is not writable: [" + resource + "]");
|
||||
state = new OutputState();
|
||||
state.setDeleteIfExists(shouldDeleteIfExists);
|
||||
state.setAppendAllowed(append);
|
||||
state.setEncoding(encoding);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encapsulates the runtime state of the writer. All state changing
|
||||
* operations on the writer go through this class.
|
||||
*/
|
||||
protected class OutputState {
|
||||
|
||||
private FileOutputStream os;
|
||||
|
||||
// The bufferedWriter over the file channel that is actually written
|
||||
Writer outputBufferedWriter;
|
||||
|
||||
FileChannel fileChannel;
|
||||
|
||||
// this represents the charset encoding (if any is needed) for the
|
||||
// output file
|
||||
String encoding = DEFAULT_CHARSET;
|
||||
|
||||
boolean restarted = false;
|
||||
|
||||
long lastMarkedByteOffsetPosition = 0;
|
||||
|
||||
long linesWritten = 0;
|
||||
|
||||
boolean shouldDeleteIfExists = true;
|
||||
|
||||
boolean initialized = false;
|
||||
|
||||
private boolean append = false;
|
||||
|
||||
private boolean appending = false;
|
||||
|
||||
/**
|
||||
* Return the byte offset position of the cursor in the output file as a
|
||||
* long integer.
|
||||
*/
|
||||
public long position() throws IOException {
|
||||
long pos = 0;
|
||||
|
||||
if (fileChannel == null) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
outputBufferedWriter.flush();
|
||||
pos = fileChannel.position();
|
||||
if (transactional) {
|
||||
pos += ((TransactionAwareBufferedWriter) outputBufferedWriter).getBufferSize();
|
||||
}
|
||||
|
||||
return pos;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param append if true, append to previously created file
|
||||
*/
|
||||
public void setAppendAllowed(boolean append) {
|
||||
this.append = append;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param executionContext state from which to restore writing from
|
||||
*/
|
||||
public void restoreFrom(ExecutionContext executionContext) {
|
||||
lastMarkedByteOffsetPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME));
|
||||
linesWritten = executionContext.getLong(getExecutionContextKey(WRITTEN_STATISTICS_NAME));
|
||||
if (shouldDeleteIfEmpty && linesWritten == 0) {
|
||||
// previous execution deleted the output file because no items were written
|
||||
restarted = false;
|
||||
lastMarkedByteOffsetPosition = 0;
|
||||
} else {
|
||||
restarted = true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param shouldDeleteIfExists indicator
|
||||
*/
|
||||
public void setDeleteIfExists(boolean shouldDeleteIfExists) {
|
||||
this.shouldDeleteIfExists = shouldDeleteIfExists;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param encoding file encoding
|
||||
*/
|
||||
public void setEncoding(String encoding) {
|
||||
this.encoding = encoding;
|
||||
}
|
||||
|
||||
public long getLinesWritten() {
|
||||
return linesWritten;
|
||||
}
|
||||
|
||||
public void setLinesWritten(long linesWritten) {
|
||||
this.linesWritten = linesWritten;
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the open resource and reset counters.
|
||||
*/
|
||||
public void close() {
|
||||
|
||||
initialized = false;
|
||||
restarted = false;
|
||||
try {
|
||||
if (outputBufferedWriter != null) {
|
||||
outputBufferedWriter.close();
|
||||
}
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
|
||||
}
|
||||
finally {
|
||||
if (!transactional) {
|
||||
closeStream();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void closeStream() {
|
||||
try {
|
||||
if (fileChannel != null) {
|
||||
fileChannel.close();
|
||||
}
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
|
||||
}
|
||||
finally {
|
||||
try {
|
||||
if (os != null) {
|
||||
os.close();
|
||||
}
|
||||
}
|
||||
catch (IOException ioe) {
|
||||
throw new ItemStreamException("Unable to close the the ItemWriter", ioe);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param line String to be written to the file
|
||||
* @throws IOException
|
||||
*/
|
||||
public void write(String line) throws IOException {
|
||||
if (!initialized) {
|
||||
initializeBufferedWriter();
|
||||
}
|
||||
|
||||
outputBufferedWriter.write(line);
|
||||
outputBufferedWriter.flush();
|
||||
}
|
||||
|
||||
/**
|
||||
* Truncate the output at the last known good point.
|
||||
*
|
||||
* @throws IOException if unable to work with file
|
||||
*/
|
||||
public void truncate() throws IOException {
|
||||
fileChannel.truncate(lastMarkedByteOffsetPosition);
|
||||
fileChannel.position(lastMarkedByteOffsetPosition);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the buffered writer for the output file channel based on
|
||||
* configuration information.
|
||||
* @throws IOException if unable to initialize buffer
|
||||
*/
|
||||
private void initializeBufferedWriter() throws IOException {
|
||||
|
||||
File file = resource.getFile();
|
||||
FileUtils.setUpOutputFile(file, restarted, append, shouldDeleteIfExists);
|
||||
|
||||
os = new FileOutputStream(file.getAbsolutePath(), true);
|
||||
fileChannel = os.getChannel();
|
||||
|
||||
outputBufferedWriter = getBufferedWriter(fileChannel, encoding);
|
||||
outputBufferedWriter.flush();
|
||||
|
||||
if (append) {
|
||||
// Bug in IO library? This doesn't work...
|
||||
// lastMarkedByteOffsetPosition = fileChannel.position();
|
||||
if (file.length() > 0) {
|
||||
appending = true;
|
||||
// Don't write the headers again
|
||||
}
|
||||
}
|
||||
|
||||
Assert.state(outputBufferedWriter != null,
|
||||
"Unable to initialize buffered writer");
|
||||
// in case of restarting reset position to last committed point
|
||||
if (restarted) {
|
||||
checkFileSize();
|
||||
truncate();
|
||||
}
|
||||
|
||||
initialized = true;
|
||||
}
|
||||
|
||||
public boolean isInitialized() {
|
||||
return initialized;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the buffered writer opened to the beginning of the file
|
||||
* specified by the absolute path name contained in absoluteFileName.
|
||||
*/
|
||||
private Writer getBufferedWriter(FileChannel fileChannel, String encoding) {
|
||||
try {
|
||||
final FileChannel channel = fileChannel;
|
||||
if (transactional) {
|
||||
TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, () -> closeStream());
|
||||
|
||||
writer.setEncoding(encoding);
|
||||
writer.setForceSync(forceSync);
|
||||
return writer;
|
||||
}
|
||||
else {
|
||||
Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) {
|
||||
@Override
|
||||
public void flush() throws IOException {
|
||||
super.flush();
|
||||
if (forceSync) {
|
||||
channel.force(false);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
return writer;
|
||||
}
|
||||
}
|
||||
catch (UnsupportedCharsetException ucse) {
|
||||
throw new ItemStreamException("Bad encoding configuration for output file " + fileChannel, ucse);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks (on setState) to make sure that the current output file's size
|
||||
* is not smaller than the last saved commit point. If it is, then the
|
||||
* file has been damaged in some way and whole task must be started over
|
||||
* again from the beginning.
|
||||
* @throws IOException if there is an IO problem
|
||||
*/
|
||||
private void checkFileSize() throws IOException {
|
||||
long size = -1;
|
||||
|
||||
outputBufferedWriter.flush();
|
||||
size = fileChannel.size();
|
||||
|
||||
if (size < lastMarkedByteOffsetPosition) {
|
||||
throw new ItemStreamException("Current file size is smaller than size at last commit");
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user