From 74364230603f06e9bece943ec7c15d9f2b12c54c Mon Sep 17 00:00:00 2001 From: robokaso Date: Fri, 7 Nov 2008 13:47:35 +0000 Subject: [PATCH] IN PROGRESS - BATCH-34: Support for multiple I/O files in a single jobRun for a particular scheduleDate. "tracer bullet" implementation --- .../batch/item/file/FlatFileItemWriter.java | 3 +- .../item/file/MultiResourceItemReader.java | 4 +- .../item/file/MultiResourceItemWriter.java | 117 ++++++++++++++++++ .../ResourceAwareItemWriterItemStream.java | 16 +++ .../batch/item/xml/StaxEventItemWriter.java | 4 +- .../file/MultiResourceItemWriterTests.java | 104 ++++++++++++++++ 6 files changed, 242 insertions(+), 6 deletions(-) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceAwareItemWriterItemStream.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterTests.java 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 4d5b6b7b7..b2515037e 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 @@ -29,7 +29,6 @@ import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.FlushFailedException; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.WriterNotOpenException; import org.springframework.batch.item.file.mapping.FieldSet; import org.springframework.batch.item.file.transform.LineAggregator; @@ -55,7 +54,7 @@ import org.springframework.util.ClassUtils; * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileItemWriter extends ExecutionContextUserSupport implements ItemWriter, ItemStream, +public class FlatFileItemWriter extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream, InitializingBean { private static final String DEFAULT_LINE_SEPARATOR = System.getProperty("line.separator"); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java index b5d18ccfa..4eb991f64 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java @@ -36,7 +36,7 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(); - private ResourceAwareItemReaderItemStream delegate; + private ResourceAwareItemReaderItemStream delegate; private Resource[] resources; @@ -160,7 +160,7 @@ public class MultiResourceItemReader implements ItemReader, ItemStream { /** * @param delegate reads items from single {@link Resource}. */ - public void setDelegate(ResourceAwareItemReaderItemStream delegate) { + public void setDelegate(ResourceAwareItemReaderItemStream delegate) { this.delegate = delegate; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java new file mode 100644 index 000000000..f2a2f71b9 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java @@ -0,0 +1,117 @@ +package org.springframework.batch.item.file; + +import java.io.File; +import java.io.IOException; +import java.util.List; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +/** + * Wraps a {@link ResourceAwareItemWriterItemStream} and creates a new output + * resource when the count of items written in current resource exceeds + * {@link #setItemCountLimitPerResource(int)}. + * + * Note that new resources are created only at chunk boundaries i.e. the number + * of items written into one resource is between the limit set by + * {@link #setItemCountLimitPerResource(int)} and (limit + chunk size). + * + * @param item type + * + * TODO strategise naming of the created resources + * + * @author Robert Kasanicky + */ +public class MultiResourceItemWriter implements ItemWriter, ItemStream { + + final static private String RESOURCE_INDEX_KEY = "resource.index"; + + final static private String CURRENT_RESOURCE_ITEM_COUNT = "resource.item.count"; + + private Resource resource; + + private ResourceAwareItemWriterItemStream delegate; + + private int itemCountLimitPerResource = Integer.MAX_VALUE; + + private int currentResourceItemCount = 0; + + private int resourceIndex = 1; + + private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport(); + + public MultiResourceItemWriter() { + setName(ClassUtils.getShortName(MultiResourceItemWriter.class)); + } + + public void write(List items) throws Exception { + if (currentResourceItemCount >= itemCountLimitPerResource) { + delegate.close(new ExecutionContext()); + resourceIndex++; + currentResourceItemCount = 0; + pointDelegateToNextResource(); + delegate.open(new ExecutionContext()); + } + delegate.write(items); + currentResourceItemCount += items.size(); + } + + public void setItemCountLimitPerResource(int itemCountLimitPerResource) { + this.itemCountLimitPerResource = itemCountLimitPerResource; + } + + public void setDelegate(ResourceAwareItemWriterItemStream delegate) { + this.delegate = delegate; + } + + public void setResource(Resource resource) { + this.resource = resource; + } + + public void close(ExecutionContext executionContext) throws ItemStreamException { + resourceIndex = 1; + currentResourceItemCount = 0; + delegate.close(executionContext); + } + + public void open(ExecutionContext executionContext) throws ItemStreamException { + resourceIndex = Long.valueOf(executionContext.getLong(ecSupport.getKey(RESOURCE_INDEX_KEY), 1L)).intValue(); + currentResourceItemCount = Long.valueOf( + executionContext.getLong(ecSupport.getKey(CURRENT_RESOURCE_ITEM_COUNT), 0L)).intValue(); + try { + pointDelegateToNextResource(); + } + catch (IOException e) { + throw new ItemStreamException("Couldn't open resource", e); + } + delegate.open(executionContext); + } + + public void update(ExecutionContext executionContext) throws ItemStreamException { + delegate.update(executionContext); + executionContext.put(ecSupport.getKey(CURRENT_RESOURCE_ITEM_COUNT), Long.valueOf(currentResourceItemCount)); + executionContext.put(ecSupport.getKey(RESOURCE_INDEX_KEY), Long.valueOf(resourceIndex)); + } + + public void setName(String name) { + ecSupport.setName(name); + } + + /** + * Create next output resource and point the delegate to it. + */ + private void pointDelegateToNextResource() throws IOException { + String path = resource.getFile().getAbsolutePath() + "." + resourceIndex; + File file = new File(path); + file.createNewFile(); + Assert.state(file.canWrite(), "Output resource " + path + " must be writable"); + delegate.setResource(new FileSystemResource(file)); + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceAwareItemWriterItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceAwareItemWriterItemStream.java new file mode 100644 index 000000000..5f37c2237 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourceAwareItemWriterItemStream.java @@ -0,0 +1,16 @@ +package org.springframework.batch.item.file; + +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; +import org.springframework.core.io.Resource; + +/** + * Interface for {@link ItemWriter}s that implement {@link ItemStream} and write + * output to {@link Resource}. + * + * @author Robert Kasanicky + */ +public interface ResourceAwareItemWriterItemStream extends ItemStream, ItemWriter { + + void setResource(Resource resource); +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index dbe577e5f..5b426bfea 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -19,9 +19,9 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.FlushFailedException; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.file.ResourceAwareItemWriterItemStream; import org.springframework.batch.item.util.ExecutionContextUserSupport; import org.springframework.batch.item.util.FileUtils; import org.springframework.batch.item.xml.stax.NoStartEndDocumentStreamWriter; @@ -49,7 +49,7 @@ import org.springframework.xml.transform.StaxResult; * @author Robert Kasanicky * */ -public class StaxEventItemWriter extends ExecutionContextUserSupport implements ItemWriter, ItemStream, +public class StaxEventItemWriter extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream, InitializingBean { private static final Log log = LogFactory.getLog(StaxEventItemWriter.class); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterTests.java new file mode 100644 index 000000000..3f15476dc --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterTests.java @@ -0,0 +1,104 @@ +package org.springframework.batch.item.file; + +import static org.junit.Assert.*; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileReader; +import java.util.Arrays; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.file.transform.PassThroughLineAggregator; +import org.springframework.core.io.FileSystemResource; + +/** + * Tests for {@link MultiResourceItemWriter}. + */ +public class MultiResourceItemWriterTests { + + private MultiResourceItemWriter tested = new MultiResourceItemWriter(); + + private File file; + + private ResourceAwareItemWriterItemStream delegate = new FlatFileItemWriter() { + { + setLineAggregator(new PassThroughLineAggregator()); + } + }; + + private ExecutionContext executionContext = new ExecutionContext(); + + @Before + public void setUp() throws Exception { + file = File.createTempFile(MultiResourceItemWriterTests.class.getSimpleName(), null); + tested.setResource(new FileSystemResource(file)); + tested.setDelegate(delegate); + tested.setItemCountLimitPerResource(2); + + tested.open(executionContext); + } + + @Test + public void testBasicMultiResourceWriteScenario() throws Exception { + + tested.write(Arrays.asList("1", "2", "3")); + + File part1 = new File(file.getAbsolutePath() + ".1"); + assertTrue(part1.exists()); + assertEquals("123", readFile(part1)); + + tested.write(Arrays.asList("4")); + File part2 = new File(file.getAbsolutePath() + ".2"); + assertTrue(part2.exists()); + assertEquals("4", readFile(part2)); + + tested.write(Arrays.asList("5")); + assertEquals("45", readFile(part2)); + + tested.write(Arrays.asList("6", "7", "8", "9")); + File part3 = new File(file.getAbsolutePath() + ".3"); + assertTrue(part3.exists()); + assertEquals("6789", readFile(part3)); + } + + @Test + public void testRestart() throws Exception { + tested.write(Arrays.asList("1", "2", "3")); + + File part1 = new File(file.getAbsolutePath() + ".1"); + assertTrue(part1.exists()); + assertEquals("123", readFile(part1)); + + tested.write(Arrays.asList("4")); + File part2 = new File(file.getAbsolutePath() + ".2"); + assertTrue(part2.exists()); + assertEquals("4", readFile(part2)); + + tested.update(executionContext); + tested.close(executionContext); + tested.open(executionContext); + + tested.write(Arrays.asList("5")); + assertEquals("45", readFile(part2)); + + tested.write(Arrays.asList("6", "7", "8", "9")); + File part3 = new File(file.getAbsolutePath() + ".3"); + assertTrue(part3.exists()); + assertEquals("6789", readFile(part3)); + } + + private String readFile(File f) throws Exception { + BufferedReader reader = new BufferedReader(new FileReader(f)); + StringBuilder result = new StringBuilder(); + while (true) { + String line = reader.readLine(); + if (line == null) { + break; + } + result.append(line); + } + return result.toString(); + } +}