diff --git a/spring-batch-docs/asciidoc/readersAndWriters.adoc b/spring-batch-docs/asciidoc/readersAndWriters.adoc index 9f601df09..1de61cf51 100644 --- a/spring-batch-docs/asciidoc/readersAndWriters.adoc +++ b/spring-batch-docs/asciidoc/readersAndWriters.adoc @@ -1748,9 +1748,9 @@ staxItemWriter.write(trade); ---- [[jsonReadingWriting]] -=== JSON Item Readers +=== JSON Item Readers And Writers -Spring Batch provides support for reading JSON resources in the following format: +Spring Batch provides support for reading and Writing JSON resources in the following format: [source, json] ---- @@ -1805,6 +1805,35 @@ public JsonItemReader jsonItemReader() { } ---- +==== `JsonFileItemWriter` + +The `JsonFileItemWriter` delegates the marshalling of items to the +`org.springframework.batch.item.json.JsonObjectMarshaller` interface. The contract +of this interface is to take an object and marshall it to a JSON `String`. +Two implementations are currently provided: + +* link:$$https://github.com/FasterXML/jackson$$[Jackson] through the `org.springframework.batch.item.json.JacksonJsonObjectMarshaller` +* link:$$https://github.com/google/gson$$[Gson] through the `org.springframework.batch.item.json.GsonJsonObjectMarshaller` + +To be able to write JSON records, the following is needed: + +* `Resource`: A Spring `Resource` that represents the JSON file to write +* `JsonObjectMarshaller`: A JSON object marshaller to marshall objects to JSON format + +The following example shows how to define a `JsonFileItemWriter`: + +[source, java] +---- +@Bean +public JsonFileItemWriter jsonFileItemWriter() { + return new JsonFileItemWriterBuilder() + .jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>()) + .resource(new ClassPathResource("trades.json")) + .name("tradeJsonFileItemWriter") + .build(); +} +---- + [[multiFileInput]] === Multi-File Input diff --git a/spring-batch-docs/asciidoc/whatsnew.adoc b/spring-batch-docs/asciidoc/whatsnew.adoc index 29bebed37..2ca1acd60 100644 --- a/spring-batch-docs/asciidoc/whatsnew.adoc +++ b/spring-batch-docs/asciidoc/whatsnew.adoc @@ -10,7 +10,7 @@ The Spring Batch 4.1 release adds the following features: * A new `@SpringBatchTest` annotation to simplify testing batch components * A new `@EnableBatchIntegration` annotation to simplify remote chunking configuration -* A new `JsonItemReader` to support the JSON format +* A new `JsonItemReader` and `JsonFileItemWriter` to support the JSON format [[whatsNewTesting]] === `@SpringBatchTest` Annotation @@ -155,6 +155,8 @@ APIs to read JSON objects in chunks. Spring Batch supports two libraries: * link:$$https://github.com/google/gson$$[Gson] To add other libraries, you can implement the `JsonObjectReader` interface. + +Writing JSON data is also supported through the `JsonFileItemWriter`. For more details about JSON support, see the <> chapter. diff --git a/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/GsonJsonFileItemWriterFunctionalTests.java b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/GsonJsonFileItemWriterFunctionalTests.java new file mode 100644 index 000000000..bc38721fb --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/GsonJsonFileItemWriterFunctionalTests.java @@ -0,0 +1,47 @@ +/* + * 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; +import com.google.gson.GsonBuilder; + +import org.springframework.batch.item.json.domain.Trade; + +/** + * @author Mahmoud Ben Hassine + */ +public class GsonJsonFileItemWriterFunctionalTests extends JsonFileItemWriterFunctionalTests { + + @Override + protected JsonObjectMarshaller getJsonObjectMarshaller() { + return new GsonJsonObjectMarshaller<>(); + } + + @Override + protected JsonObjectMarshaller getJsonObjectMarshallerWithPrettyPrint() { + Gson gson = new GsonBuilder().setPrettyPrinting().create(); + GsonJsonObjectMarshaller jsonObjectMarshaller = new GsonJsonObjectMarshaller<>(); + jsonObjectMarshaller.setGson(gson); + return jsonObjectMarshaller; + } + + @Override + protected String getExpectedPrettyPrintedFile() { + return "expected-trades-gson-pretty-print.json"; + } + +} diff --git a/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/JacksonJsonFileItemWriterFunctionalTests.java b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/JacksonJsonFileItemWriterFunctionalTests.java new file mode 100644 index 000000000..20b24683f --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/JacksonJsonFileItemWriterFunctionalTests.java @@ -0,0 +1,48 @@ +/* + * 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.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; + +import org.springframework.batch.item.json.domain.Trade; + +/** + * @author Mahmoud Ben Hassine + */ +public class JacksonJsonFileItemWriterFunctionalTests extends JsonFileItemWriterFunctionalTests { + + @Override + protected JsonObjectMarshaller getJsonObjectMarshaller() { + return new JacksonJsonObjectMarshaller<>(); + } + + @Override + protected JsonObjectMarshaller getJsonObjectMarshallerWithPrettyPrint() { + ObjectMapper objectMapper = new ObjectMapper(); + objectMapper.enable(SerializationFeature.INDENT_OUTPUT); + JacksonJsonObjectMarshaller jsonObjectMarshaller = new JacksonJsonObjectMarshaller<>(); + jsonObjectMarshaller.setObjectMapper(objectMapper); + return jsonObjectMarshaller; + } + + @Override + protected String getExpectedPrettyPrintedFile() { + return "expected-trades-jackson-pretty-print.json"; + } + +} diff --git a/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/JsonFileItemWriterFunctionalTests.java b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/JsonFileItemWriterFunctionalTests.java new file mode 100644 index 000000000..3146c56cc --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/item/json/JsonFileItemWriterFunctionalTests.java @@ -0,0 +1,295 @@ +/* + * 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.io.File; +import java.io.FileInputStream; +import java.math.BigDecimal; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.batch.item.json.builder.JsonFileItemWriterBuilder; +import org.springframework.batch.item.json.domain.Trade; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.util.DigestUtils; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * @author Mahmoud Ben Hassine + */ +public abstract class JsonFileItemWriterFunctionalTests { + + private static final String EXPECTED_FILE_DIRECTORY = "src/test/resources/org/springframework/batch/item/json/"; + + private Resource resource; + private List items; + private ExecutionContext executionContext; + private Trade trade1 = new Trade("123", 5, new BigDecimal("10.5"), "foo"); + private Trade trade2 = new Trade("456", 10, new BigDecimal("20.5"), "bar"); + + protected abstract JsonObjectMarshaller getJsonObjectMarshaller(); + protected abstract JsonObjectMarshaller getJsonObjectMarshallerWithPrettyPrint(); + protected abstract String getExpectedPrettyPrintedFile(); + + private JsonFileItemWriter writer; + + @Before + public void setUp() throws Exception { + Path outputFilePath = Paths.get("build", "trades.json"); + Files.deleteIfExists(outputFilePath); + this.resource = new FileSystemResource(outputFilePath.toFile()); + this.executionContext = new ExecutionContext(); + this.items = Arrays.asList(this.trade1, this.trade2); + this.writer = new JsonFileItemWriterBuilder() + .name("tradesItemWriter") + .resource(this.resource) + .jsonObjectMarshaller(getJsonObjectMarshaller()) + .build(); + } + + @Test + public void testJsonWriting() throws Exception { + // when + this.writer.open(this.executionContext); + this.writer.write(this.items); + this.writer.close(); + + // then + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + "expected-trades.json"), + this.resource.getFile()); + } + + @Test + public void testJsonWritingWithPrettyPrinting() throws Exception { + // given + this.writer = new JsonFileItemWriterBuilder() + .name("tradesItemWriter") + .resource(this.resource) + .jsonObjectMarshaller(getJsonObjectMarshallerWithPrettyPrint()) + .build(); + + // when + this.writer.open(this.executionContext); + this.writer.write(this.items); + this.writer.close(); + + // when + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + getExpectedPrettyPrintedFile()), + this.resource.getFile()); + } + + @Test + public void testJsonWritingWithEnclosingObject() throws Exception { + // given + this.writer.setHeaderCallback(writer -> writer.write("{\"trades\":[")); + this.writer.setFooterCallback(writer -> writer.write(JsonFileItemWriter.DEFAULT_LINE_SEPARATOR + "]}")); + + // when + this.writer.open(this.executionContext); + this.writer.write(this.items); + this.writer.close(); + + // then + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + "expected-trades-with-wrapper-object.json"), + this.resource.getFile()); + } + + @Test + public void testForcedWrite() throws Exception { + // given + this.writer.setForceSync(true); + + // when + this.writer.open(this.executionContext); + this.writer.write(Collections.singletonList(this.trade1)); + this.writer.close(); + + // then + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + "expected-trades1.json"), + this.resource.getFile()); + } + + @Test + public void testWriteWithDelete() throws Exception { + // given + this.writer.setShouldDeleteIfExists(true); + + // when + this.writer.open(this.executionContext); + this.writer.write(Collections.singletonList(this.trade1)); + this.writer.close(); + this.writer.open(this.executionContext); + this.writer.write(Collections.singletonList(this.trade2)); + this.writer.close(); + + // then + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + "expected-trades2.json"), + this.resource.getFile()); + } + + @Test + public void testRestart() throws Exception { + this.writer.open(this.executionContext); + // write some lines + this.writer.write(Collections.singletonList(this.trade1)); + // get restart data + this.writer.update(this.executionContext); + // close template + this.writer.close(); + + // init with correct data + this.writer.open(this.executionContext); + // write more lines + this.writer.write(Collections.singletonList(this.trade2)); + // get statistics + this.writer.update(this.executionContext); + // close template + this.writer.close(); + + // verify what was written to the file + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY+ "expected-trades.json"), + this.resource.getFile()); + + // 2 lines were written to the file in total + assertEquals(2, this.executionContext.getLong("tradesItemWriter.written")); + } + + @Test + public void testTransactionalRestart() throws Exception { + this.writer.open(this.executionContext); + PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); + + new TransactionTemplate(transactionManager).execute((TransactionCallback) status -> { + try { + // write some lines + this.writer.write(Collections.singletonList(this.trade1)); + } + catch (Exception e) { + throw new UnexpectedInputException("Could not write data", e); + } + // get restart data + this.writer.update(this.executionContext); + return null; + }); + // close template + this.writer.close(); + + // init with correct data + this.writer.open(this.executionContext); + + new TransactionTemplate(transactionManager).execute((TransactionCallback) status -> { + try { + // write more lines + this.writer.write(Collections.singletonList(this.trade2)); + } + catch (Exception e) { + throw new UnexpectedInputException("Could not write data", e); + } + // get restart data + this.writer.update(this.executionContext); + return null; + }); + // close template + this.writer.close(); + + // verify what was written to the file + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY+ "expected-trades.json"), + this.resource.getFile()); + + // 2 lines were written to the file in total + assertEquals(2, this.executionContext.getLong("tradesItemWriter.written")); + } + + @Test + public void testItemMarshallingFailure() throws Exception { + this.writer.setJsonObjectMarshaller(item -> { + throw new IllegalArgumentException("Bad item"); + }); + this.writer.open(this.executionContext); + try { + this.writer.write(Collections.singletonList(this.trade1)); + fail(); + } + catch (IllegalArgumentException iae) { + assertEquals("Bad item", iae.getMessage()); + } + finally { + this.writer.close(); + } + + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + "empty-trades.json"), + this.resource.getFile()); + } + + @Test + /* + * If append=true a new output file should still be created on the first run (not restart). + */ + public void testAppendToNotYetExistingFile() throws Exception { + Resource toBeCreated = new FileSystemResource("build/FlatFileItemWriterTests.out"); + + File outputFile = toBeCreated.getFile(); //enable easy content reading and auto-delete the file + + assertFalse("output file does not exist yet", toBeCreated.exists()); + this.writer.setResource(toBeCreated); + this.writer.setAppendAllowed(true); + this.writer.afterPropertiesSet(); + + this.writer.open(this.executionContext); + assertTrue("output file was created", toBeCreated.exists()); + + this.writer.write(Collections.singletonList(this.trade1)); + this.writer.close(); + assertFileEquals( + new File(EXPECTED_FILE_DIRECTORY + "expected-trades1.json"), + outputFile); + outputFile.delete(); + } + + private void assertFileEquals(File expected, File actual) throws Exception { + String expectedHash = DigestUtils.md5DigestAsHex(new FileInputStream(expected)); + String actualHash = DigestUtils.md5DigestAsHex(new FileInputStream(actual)); + Assert.assertEquals(expectedHash, actualHash); + } +} diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/empty-trades.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/empty-trades.json new file mode 100644 index 000000000..41b42e677 --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/empty-trades.json @@ -0,0 +1,3 @@ +[ + +] diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-gson-pretty-print.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-gson-pretty-print.json new file mode 100644 index 000000000..51fb2644e --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-gson-pretty-print.json @@ -0,0 +1,14 @@ +[ + { + "isin": "123", + "quantity": 5, + "price": 10.5, + "customer": "foo" +}, + { + "isin": "456", + "quantity": 10, + "price": 20.5, + "customer": "bar" +} +] diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-jackson-pretty-print.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-jackson-pretty-print.json new file mode 100644 index 000000000..36a470f62 --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-jackson-pretty-print.json @@ -0,0 +1,14 @@ +[ + { + "isin" : "123", + "quantity" : 5, + "price" : 10.5, + "customer" : "foo" +}, + { + "isin" : "456", + "quantity" : 10, + "price" : 20.5, + "customer" : "bar" +} +] diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-with-wrapper-object.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-with-wrapper-object.json new file mode 100644 index 000000000..e8cfeca82 --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades-with-wrapper-object.json @@ -0,0 +1,4 @@ +{"trades":[ + {"isin":"123","quantity":5,"price":10.5,"customer":"foo"}, + {"isin":"456","quantity":10,"price":20.5,"customer":"bar"} +]} \ No newline at end of file diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades.json new file mode 100644 index 000000000..49a4b7bcb --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades.json @@ -0,0 +1,4 @@ +[ + {"isin":"123","quantity":5,"price":10.5,"customer":"foo"}, + {"isin":"456","quantity":10,"price":20.5,"customer":"bar"} +] diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades1.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades1.json new file mode 100644 index 000000000..ae3c0b08b --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades1.json @@ -0,0 +1,3 @@ +[ + {"isin":"123","quantity":5,"price":10.5,"customer":"foo"} +] diff --git a/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades2.json b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades2.json new file mode 100644 index 000000000..e7ac19903 --- /dev/null +++ b/spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/item/json/expected-trades2.json @@ -0,0 +1,3 @@ +[ + {"isin":"456","quantity":10,"price":20.5,"customer":"bar"} +] diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index 5b82b06e8..610c34675 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -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 extends AbstractItemStreamItemWriter implements ResourceAwareItemWriterItemStream, -InitializingBean { +public class FlatFileItemWriter extends AbstractFileItemWriter { - 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 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 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.
- * - * @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 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 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(); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/GsonJsonObjectMarshaller.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/GsonJsonObjectMarshaller.java new file mode 100644 index 000000000..a070d6671 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/GsonJsonObjectMarshaller.java @@ -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 Google Gson + * to marshal an object into a json representation. + * + * @param type of objects to marshal + * @author Mahmoud Ben Hassine + * @since 4.1 + */ +public class GsonJsonObjectMarshaller implements JsonObjectMarshaller { + + 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); + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JacksonJsonObjectMarshaller.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JacksonJsonObjectMarshaller.java new file mode 100644 index 000000000..7175d97e7 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JacksonJsonObjectMarshaller.java @@ -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 Jackson + * to marshal an object into a json representation. + * + * @param type of objects to marshal + * @author Mahmoud Ben Hassine + * @since 4.1 + */ +public class JacksonJsonObjectMarshaller implements JsonObjectMarshaller { + + 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); + } + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonFileItemWriter.java new file mode 100644 index 000000000..39c26e748 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonFileItemWriter.java @@ -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: + * + *

+ * + * [ + * {json object}, + * {json object}, + * {json object} + * ] + * + *

+ * + * The implementation is not thread-safe. + * + * @see GsonJsonObjectMarshaller + * @see JacksonJsonObjectMarshaller + * @param type of object to write as json representation + * @author Mahmoud Ben Hassine + * @since 4.1 + */ +public class JsonFileItemWriter extends AbstractFileItemWriter { + + private static final char JSON_OBJECT_SEPARATOR = ','; + private static final char JSON_ARRAY_START = '['; + private static final char JSON_ARRAY_STOP = ']'; + + private JsonObjectMarshaller 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 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 jsonObjectMarshaller) { + this.jsonObjectMarshaller = jsonObjectMarshaller; + } + + @Override + public String doWrite(List items) { + StringBuilder lines = new StringBuilder(); + Iterator 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(); + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonObjectMarshaller.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonObjectMarshaller.java new file mode 100644 index 000000000..73923ca56 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonObjectMarshaller.java @@ -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 type of objects to marshal + * @author Mahmoud Ben Hassine + * @since 4.1 + */ +public interface JsonObjectMarshaller { + + /** + * Marshal an object into a json representation. + * @param object to marshal + * @return json representation fo the object + */ + String marshal(T object); + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/builder/JsonFileItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/builder/JsonFileItemWriterBuilder.java new file mode 100644 index 000000000..0dd283a4e --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/builder/JsonFileItemWriterBuilder.java @@ -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 type of objects to write as Json output. + * @author Mahmoud Ben Hassine + * @since 4.1 + */ +public class JsonFileItemWriterBuilder { + + private Resource resource; + private JsonObjectMarshaller 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 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 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 forceSync(boolean forceSync) { + this.forceSync = forceSync; + + return this; + } + + /** + * String used to separate lines in output. Defaults to the System property + * line.separator. + * + * @param lineSeparator value to use for a line separator + * @return The current instance of the builder. + * @see JsonFileItemWriter#setLineSeparator(String) + */ + public JsonFileItemWriterBuilder 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 jsonObjectMarshaller(JsonObjectMarshaller 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 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 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 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 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 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 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 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 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 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 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; + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractFileItemWriter.java new file mode 100644 index 000000000..27a517ed1 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractFileItemWriter.java @@ -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.
+ * + * Uses buffered writer to improve performance.
+ * + * The implementation is not 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 extends AbstractItemStreamItemWriter + implements ResourceAwareItemWriterItemStream, 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 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 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"); + } + } + + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/GsonJsonObjectMarshallerTest.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/GsonJsonObjectMarshallerTest.java new file mode 100644 index 000000000..cade71dfd --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/GsonJsonObjectMarshallerTest.java @@ -0,0 +1,65 @@ +/* + * 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 org.junit.Assert; +import org.junit.Test; + +/** + * @author Mahmoud Ben Hassine + */ +public class GsonJsonObjectMarshallerTest { + + @Test + public void testJsonMarshalling() { + // given + GsonJsonObjectMarshaller jsonObjectMarshaller = new GsonJsonObjectMarshaller<>(); + + // when + String foo = jsonObjectMarshaller.marshal(new Foo(1, "foo")); + + // then + Assert.assertEquals("{\"id\":1,\"name\":\"foo\"}", foo); + } + + public static class Foo { + private int id; + private String name; + + public Foo(int id, String name) { + this.id = id; + this.name = name; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/JacksonJsonObjectMarshallerTest.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/JacksonJsonObjectMarshallerTest.java new file mode 100644 index 000000000..687e38b35 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/JacksonJsonObjectMarshallerTest.java @@ -0,0 +1,65 @@ +/* + * 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 org.junit.Assert; +import org.junit.Test; + +/** + * @author Mahmoud Ben Hassine + */ +public class JacksonJsonObjectMarshallerTest { + + @Test + public void testJsonMarshalling() { + // given + JacksonJsonObjectMarshaller jsonObjectMarshaller = new JacksonJsonObjectMarshaller<>(); + + // when + String foo = jsonObjectMarshaller.marshal(new Foo(1, "foo")); + + // then + Assert.assertEquals("{\"id\":1,\"name\":\"foo\"}", foo); + } + + public static class Foo { + private int id; + private String name; + + public Foo(int id, String name) { + this.id = id; + this.name = name; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/JsonFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/JsonFileItemWriterTests.java new file mode 100644 index 000000000..d6016612d --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/JsonFileItemWriterTests.java @@ -0,0 +1,70 @@ +/* + * 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.io.File; +import java.nio.file.Files; +import java.util.Arrays; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; + +/** + * @author Mahmoud Ben Hassine + */ +public class JsonFileItemWriterTests { + + private Resource resource; + private JsonObjectMarshaller jsonObjectMarshaller; + + @Before + public void setUp() throws Exception { + File file = Files.createTempFile("test", "json").toFile(); + this.resource = new FileSystemResource(file); + this.jsonObjectMarshaller = Mockito.mock(JsonObjectMarshaller.class); + } + + @Test(expected = IllegalArgumentException.class) + public void resourceMustNotBeNull() { + new JsonFileItemWriter<>(null, this.jsonObjectMarshaller); + } + + @Test(expected = IllegalArgumentException.class) + public void jsonObjectMarshallerMustNotBeNull() { + new JsonFileItemWriter<>(this.resource, null); + } + + @Test + public void itemsShouldBeMarshalledToJsonWithTheJsonObjectMarshaller() throws Exception { + // given + JsonFileItemWriter writer = new JsonFileItemWriter<>(this.resource, this.jsonObjectMarshaller); + + // when + writer.open(new ExecutionContext()); + writer.write(Arrays.asList("foo", "bar")); + writer.close(); + + // then + Mockito.verify(this.jsonObjectMarshaller).marshal("foo"); + Mockito.verify(this.jsonObjectMarshaller).marshal("bar"); + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/builder/JsonFileItemWriterBuilderTest.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/builder/JsonFileItemWriterBuilderTest.java new file mode 100644 index 000000000..061186f0b --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/json/builder/JsonFileItemWriterBuilderTest.java @@ -0,0 +1,116 @@ +/* + * 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 java.io.File; +import java.nio.file.Files; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +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.FileSystemResource; +import org.springframework.core.io.Resource; +import org.springframework.test.util.ReflectionTestUtils; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * @author Mahmoud Ben Hassine + */ +public class JsonFileItemWriterBuilderTest { + + private Resource resource; + private JsonObjectMarshaller jsonObjectMarshaller; + + @Before + public void setUp() throws Exception { + File file = Files.createTempFile("test", "json").toFile(); + this.resource = new FileSystemResource(file); + this.jsonObjectMarshaller = object -> object; + } + + @Test(expected = IllegalArgumentException.class) + public void testMissingResource() { + new JsonFileItemWriterBuilder() + .jsonObjectMarshaller(this.jsonObjectMarshaller) + .build(); + } + + @Test(expected = IllegalArgumentException.class) + public void testMissingJsonObjectMarshaller() { + new JsonFileItemWriterBuilder() + .resource(this.resource) + .build(); + } + + @Test(expected = IllegalArgumentException.class) + public void testMandatoryNameWhenSaveStateIsSet() { + new JsonFileItemWriterBuilder() + .resource(this.resource) + .jsonObjectMarshaller(this.jsonObjectMarshaller) + .build(); + } + + @Test + public void testJsonFileItemWriterCreation() { + // given + boolean append = true; + boolean forceSync = true; + boolean transactional = true; + boolean shouldDeleteIfEmpty = true; + boolean shouldDeleteIfExists = true; + String encoding = "UTF-8"; + String lineSeparator = "#"; + FlatFileHeaderCallback headerCallback = Mockito.mock(FlatFileHeaderCallback.class); + FlatFileFooterCallback footerCallback = Mockito.mock(FlatFileFooterCallback.class); + + // when + JsonFileItemWriter writer = new JsonFileItemWriterBuilder() + .name("jsonFileItemWriter") + .resource(this.resource) + .jsonObjectMarshaller(this.jsonObjectMarshaller) + .append(append) + .encoding(encoding) + .forceSync(forceSync) + .headerCallback(headerCallback) + .footerCallback(footerCallback) + .lineSeparator(lineSeparator) + .shouldDeleteIfEmpty(shouldDeleteIfEmpty) + .shouldDeleteIfExists(shouldDeleteIfExists) + .transactional(transactional) + .build(); + + // then + assertTrue((Boolean) ReflectionTestUtils.getField(writer, "saveState")); + assertTrue((Boolean) ReflectionTestUtils.getField(writer, "append")); + assertTrue((Boolean) ReflectionTestUtils.getField(writer, "transactional")); + assertTrue((Boolean) ReflectionTestUtils.getField(writer, "shouldDeleteIfEmpty")); + assertTrue((Boolean) ReflectionTestUtils.getField(writer, "shouldDeleteIfExists")); + assertTrue((Boolean) ReflectionTestUtils.getField(writer, "forceSync")); + assertEquals(encoding, ReflectionTestUtils.getField(writer, "encoding")); + assertEquals(lineSeparator, ReflectionTestUtils.getField(writer, "lineSeparator")); + assertEquals(headerCallback, ReflectionTestUtils.getField(writer, "headerCallback")); + assertEquals(footerCallback, ReflectionTestUtils.getField(writer, "footerCallback")); + } + +} diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/JsonSupportIntegrationTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/JsonSupportIntegrationTests.java new file mode 100644 index 000000000..6a289ee46 --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/JsonSupportIntegrationTests.java @@ -0,0 +1,129 @@ +/* + * 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.sample; + +import java.io.File; +import java.io.FileInputStream; +import java.nio.file.Files; +import java.nio.file.Paths; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.item.json.GsonJsonObjectReader; +import org.springframework.batch.item.json.JacksonJsonObjectMarshaller; +import org.springframework.batch.item.json.JsonItemReader; +import org.springframework.batch.item.json.JsonFileItemWriter; +import org.springframework.batch.item.json.builder.JsonItemReaderBuilder; +import org.springframework.batch.item.json.builder.JsonFileItemWriterBuilder; +import org.springframework.batch.sample.domain.trade.Trade; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.FileSystemResource; +import org.springframework.util.DigestUtils; + +/** + * @author Mahmoud Ben Hassine + */ +public class JsonSupportIntegrationTests { + + private static final String INPUT_FILE_DIRECTORY = "src/test/resources/org/springframework/batch/item/json/"; + private static final String OUTPUT_FILE_DIRECTORY = "build/"; + + @Before + public void setUp() throws Exception { + Files.deleteIfExists(Paths.get("build", "trades.json")); + } + + @Configuration + @EnableBatchProcessing + public static class JobConfiguration { + + @Autowired + private JobBuilderFactory jobs; + + @Autowired + private StepBuilderFactory steps; + + @Bean + public JsonItemReader itemReader() { + return new JsonItemReaderBuilder() + .name("tradesJsonItemReader") + .resource(new FileSystemResource(INPUT_FILE_DIRECTORY + "trades.json")) + .jsonObjectReader(new GsonJsonObjectReader<>(Trade.class)) + .build(); + } + + @Bean + public JsonFileItemWriter itemWriter() { + return new JsonFileItemWriterBuilder() + .resource(new FileSystemResource(OUTPUT_FILE_DIRECTORY + "trades.json")) + .jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>()) + .name("tradesJsonFileItemWriter") + .build(); + } + + @Bean + public Step step() { + return steps.get("step") + .chunk(2) + .reader(itemReader()) + .writer(itemWriter()) + .build(); + } + + @Bean + public Job job() { + return jobs.get("job") + .start(step()) + .build(); + } + } + + @Test + public void testJsonReadingAndWriting() throws Exception { + ApplicationContext context = new AnnotationConfigApplicationContext(JobConfiguration.class); + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + Job job = context.getBean(Job.class); + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + assertFileEquals( + new File(INPUT_FILE_DIRECTORY + "trades.json"), + new File(OUTPUT_FILE_DIRECTORY + "trades.json")); + } + + private void assertFileEquals(File expected, File actual) throws Exception { + String expectedHash = DigestUtils.md5DigestAsHex(new FileInputStream(expected)); + String actualHash = DigestUtils.md5DigestAsHex(new FileInputStream(actual)); + Assert.assertEquals(expectedHash, actualHash); + } + +} diff --git a/spring-batch-samples/src/test/resources/org/springframework/batch/item/json/trades.json b/spring-batch-samples/src/test/resources/org/springframework/batch/item/json/trades.json new file mode 100644 index 000000000..d5ebcef26 --- /dev/null +++ b/spring-batch-samples/src/test/resources/org/springframework/batch/item/json/trades.json @@ -0,0 +1,5 @@ +[ + {"isin":"123","quantity":5,"price":10.5,"customer":"foo","id":1,"version":0}, + {"isin":"456","quantity":10,"price":20.5,"customer":"bar","id":2,"version":0}, + {"isin":"789","quantity":15,"price":30.5,"customer":"baz","id":3,"version":0} +]