This class will be updated in the future to use a buffering approach - * to handling transactions, rather than outputting directly to the file and - * truncating on rollback
+ *+ * This class will be updated in the future to use a buffering approach to + * handling transactions, rather than outputting directly to the file and + * truncating on rollback + *
* * @author Waseem Malik * @author Tomas Slanina @@ -78,26 +82,40 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements private OutputState state = null; - private ItemTransformer transformer = new RecursiveCollectionItemTransformer(); + private LineAggregator lineAggregator = new DelimitedLineAggregator(); + + private FieldSetUnmapper fieldSetUnmapper; /** * Assert that mandatory properties (resource) are set. * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { - Assert.notNull(resource); + Assert.notNull(resource, "The resource must be set"); + Assert.notNull(fieldSetUnmapper, "A FieldSetUnmapper must be provided."); File file = resource.getFile(); Assert.state(!file.exists() || file.canWrite(), "Resource is not writable: [" + resource + "]"); } /** - * Public setter for the converter. If not-null this will be used to convert - * the input data before it is output. + * Public setter for the {@link LineAggregator}. This will be used to + * translate a {@link FieldSet} into a line for output. * - * @param transformer the converter to set + * @param lineAggregator the {@link LineAggregator} to set */ - public void setTransformer(ItemTransformer transformer) { - this.transformer = transformer; + public void setLineAggregator(LineAggregator lineAggregator) { + this.lineAggregator = lineAggregator; + } + + /** + * Public setter for the {@link FieldSetUnmapper}. This will be used to + * transform the item into a {@link FieldSet} before it is aggregated by the + * {@link LineAggregator}. + * + * @param fieldSetUnmapper the {@link FieldSetUnmapper} to set + */ + public void setFieldSetUnmapper(FieldSetUnmapper fieldSetUnmapper) { + this.fieldSetUnmapper = fieldSetUnmapper; } /** @@ -123,7 +141,8 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements * @throws Exception if the transformer or file output fail */ public void write(Object data) throws Exception { - getOutputState().write(transformer.transform(data) + LINE_SEPARATOR); + FieldSet fieldSet = fieldSetUnmapper.unmapItem(data); + getOutputState().write(lineAggregator.aggregate(fieldSet ) + LINE_SEPARATOR); } /** @@ -228,6 +247,13 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements long restartCount = 0; boolean shouldDeleteIfExists = true; + + /** + * + */ + public OutputState() { + initializeBufferedWriter(); + } /** * Return the byte offset position of the cursor in the output file as a @@ -469,16 +495,16 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements return true; } - /* To be deleted once interface changes are complete - * (non-Javadoc) + /* + * To be deleted once interface changes are complete (non-Javadoc) * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionContext) */ public void mark() { - + } - /* To be deleted once interface changes are complete - * (non-Javadoc) + /* + * To be deleted once interface changes are complete (non-Javadoc) * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionContext) */ public void reset() throws ResetFailedException { @@ -488,7 +514,8 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements public void clear() throws Exception { try { getOutputState().reset(); - } catch (BatchCriticalException e) { + } + catch (BatchCriticalException e) { throw new ResetFailedException(e); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/FieldSetUnmapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/FieldSetUnmapper.java new file mode 100644 index 000000000..680424cb1 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/FieldSetUnmapper.java @@ -0,0 +1,31 @@ +/* + * Copyright 2006-2007 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.io.file.mapping; + + +/** + * @author Dave Syer + * + */ +public interface FieldSetUnmapper { + + /** + * @param data an Object to convert. + * @return a {@link FieldSet} created from the input. + */ + FieldSet unmapItem(Object data); + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/PassThroughFieldSetMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/PassThroughFieldSetMapper.java index 09f3fc151..d8155e468 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/PassThroughFieldSetMapper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/mapping/PassThroughFieldSetMapper.java @@ -17,20 +17,37 @@ package org.springframework.batch.io.file.mapping; /** - * Pass through {@link FieldSetMapper} useful for - * passing a fieldset back from a FlatFileInputsource rather - * than a mapped object. - * + * Pass through {@link FieldSetMapper} useful for passing a {@link FieldSet} + * back directly rather than a mapped object. + * * @author Lucas Ward - * + * */ -public class PassThroughFieldSetMapper implements FieldSetMapper { +public class PassThroughFieldSetMapper implements FieldSetMapper, FieldSetUnmapper { - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.batch.io.file.FieldSetMapper#mapLine(org.springframework.batch.io.file.FieldSet) */ public Object mapLine(FieldSet fs) { return fs; } + /** + * If the input is a {@link FieldSet} pass it to the caller. Otherwise + * convert to a String with toString() and convert it to a single field + * {@link FieldSet}. + * + * @see org.springframework.batch.io.file.mapping.FieldSetUnmapper#unmapItem(java.lang.Object) + */ + public FieldSet unmapItem(Object data) { + if (data instanceof FieldSet) { + return (FieldSet) data; + } + if (!(data instanceof String)) { + data = "" + data; + } + return new DefaultFieldSet(new String[] { (String) data }); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/transform/FieldSetUnmapperItemTransformer.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/transform/FieldSetUnmapperItemTransformer.java new file mode 100644 index 000000000..3667c239d --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/transform/FieldSetUnmapperItemTransformer.java @@ -0,0 +1,49 @@ +/* + * Copyright 2006-2007 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.io.file.transform; + +import org.springframework.batch.io.file.mapping.FieldSet; +import org.springframework.batch.io.file.mapping.FieldSetUnmapper; +import org.springframework.batch.item.writer.ItemTransformer; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; + +/** + * An {@link ItemTransformer} that delegates to a {@link FieldSetUnmapper}, so + * the result of the transformation is a {@link FieldSet}. + * + * @author Dave Syer + * + */ +public class FieldSetUnmapperItemTransformer implements ItemTransformer, InitializingBean { + + private FieldSetUnmapper fieldSetUnmapper; + + /** + * Assert that mandatory properties are set. + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(fieldSetUnmapper, "A FieldSetUnmapper must be provided."); + } + + /* (non-Javadoc) + * @see org.springframework.batch.item.writer.ItemTransformer#transform(java.lang.Object) + */ + public Object transform(Object item) throws Exception { + return fieldSetUnmapper.unmapItem(item); + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java index 77385fdda..d43153f6c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java @@ -1,6 +1,10 @@ package org.springframework.batch.item.writer; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.exception.StreamException; +import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -10,9 +14,11 @@ import org.springframework.util.Assert; * @author Dave Syer * @author Robert Kasanicky */ -public class DelegatingItemWriter implements ItemWriter, InitializingBean { +public class DelegatingItemWriter implements ItemWriter, ItemStream, InitializingBean { private ItemWriter writer; + + private ItemStream stream; /** * Calls {@link #doProcess(Object)} and then writes the result to the @@ -40,6 +46,11 @@ public class DelegatingItemWriter implements ItemWriter, InitializingBean { */ public void setDelegate(ItemWriter writer) { this.writer = writer; + if (writer instanceof ItemStream) { + this.stream = (ItemStream) writer; + } else { + this.stream = new ItemStreamAdapter(); + } } public void afterPropertiesSet() throws Exception { @@ -60,4 +71,38 @@ public class DelegatingItemWriter implements ItemWriter, InitializingBean { writer.flush(); } + /** + * @throws StreamException + * @see org.springframework.batch.item.ItemStream#close() + */ + public void close() throws StreamException { + stream.close(); + } + + /** + * @return + * @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext() + */ + public ExecutionContext getExecutionContext() { + return stream.getExecutionContext(); + } + + /** + * @throws StreamException + * @see org.springframework.batch.item.ItemStream#open() + */ + public void open() throws StreamException { + stream.open(); + } + + /** + * @param context + * @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionContext) + */ + public void restoreFrom(ExecutionContext context) { + stream.restoreFrom(context); + } + + + } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java index f31f93533..1f9b7c402 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java @@ -20,12 +20,14 @@ import java.io.BufferedReader; import java.io.File; import java.io.FileReader; import java.io.IOException; -import java.util.Collections; import junit.framework.TestCase; +import org.springframework.batch.io.file.mapping.DefaultFieldSet; +import org.springframework.batch.io.file.mapping.FieldSet; +import org.springframework.batch.io.file.mapping.FieldSetUnmapper; +import org.springframework.batch.io.file.mapping.PassThroughFieldSetMapper; import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.writer.ItemTransformer; import org.springframework.core.io.FileSystemResource; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -66,6 +68,7 @@ public class FlatFileItemWriterTests extends TestCase { outputFile = File.createTempFile("flatfile-output-", ".tmp"); inputSource.setResource(new FileSystemResource(outputFile)); + inputSource.setFieldSetUnmapper(new PassThroughFieldSetMapper()); inputSource.afterPropertiesSet(); inputSource.open(); @@ -99,7 +102,7 @@ public class FlatFileItemWriterTests extends TestCase { /** * Regular usage ofwrite(String) method
- * @throws Exception
+ * @throws Exception
*/
public void testWriteString() throws Exception {
inputSource.write(TEST_STRING);
@@ -111,23 +114,12 @@ public class FlatFileItemWriterTests extends TestCase {
/**
* Regular usage of write(String) method
- * @throws Exception
- */
- public void testWriteCollection() throws Exception {
- inputSource.write(Collections.singleton(TEST_STRING));
- inputSource.close();
- String lineFromFile = readLine();
- assertEquals(TEST_STRING, lineFromFile);
- }
-
- /**
- * Regular usage of write(String) method
- * @throws Exception
+ * @throws Exception
*/
public void testWriteWithConverter() throws Exception {
- inputSource.setTransformer(new ItemTransformer() {
- public Object transform(Object input) {
- return "FOO:" + input;
+ inputSource.setFieldSetUnmapper(new FieldSetUnmapper() {
+ public FieldSet unmapItem(Object data) {
+ return new DefaultFieldSet(new String[] { "FOO:" + data });
}
});
Object data = new Object();
@@ -140,12 +132,12 @@ public class FlatFileItemWriterTests extends TestCase {
/**
* Regular usage of write(String) method
- * @throws Exception
+ * @throws Exception
*/
public void testWriteWithConverterAndInfiniteLoop() throws Exception {
- inputSource.setTransformer(new ItemTransformer() {
- public Object transform(Object input) {
- return "FOO:" + input;
+ inputSource.setFieldSetUnmapper(new FieldSetUnmapper() {
+ public FieldSet unmapItem(Object data) {
+ return new DefaultFieldSet(new String[] { "FOO:" + data });
}
});
Object data = new Object();
@@ -158,36 +150,23 @@ public class FlatFileItemWriterTests extends TestCase {
/**
* Regular usage of write(String) method
- * @throws Exception
+ * @throws Exception
*/
public void testWriteWithConverterAndString() throws Exception {
- inputSource.setTransformer(new ItemTransformer() {
- public Object transform(Object input) {
- return "FOO:" + input;
+ inputSource.setFieldSetUnmapper(new FieldSetUnmapper() {
+ public FieldSet unmapItem(Object data) {
+ return new DefaultFieldSet(new String[] { "FOO:" + data });
}
});
inputSource.write(TEST_STRING);
inputSource.close();
String lineFromFile = readLine();
- assertEquals("FOO:"+TEST_STRING, lineFromFile);
- }
-
- /**
- * Regular usage of write(String) method
- * @throws Exception
- */
- public void testWriteArray() throws Exception {
- inputSource.write(new String[] { TEST_STRING, TEST_STRING });
- inputSource.close();
- String lineFromFile = readLine();
- assertEquals(TEST_STRING, lineFromFile);
- lineFromFile = readLine();
- assertEquals(TEST_STRING, lineFromFile);
+ assertEquals("FOO:" + TEST_STRING, lineFromFile);
}
/**
* Regular usage of write(String[], LineDescriptor) method
- * @throws Exception
+ * @throws Exception
*/
public void testWriteRecord() throws Exception {
String args = "1";
@@ -301,11 +280,11 @@ public class FlatFileItemWriterTests extends TestCase {
assertEquals(0, streamContext.getLong(FlatFileItemWriter.RESTART_DATA_NAME));
}
- private void commit() throws Exception{
+ private void commit() throws Exception {
inputSource.flush();
}
- private void rollback() throws Exception{
+ private void rollback() throws Exception {
inputSource.clear();
}
diff --git a/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml b/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml
index ed1b46ff7..71d752174 100644
--- a/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml
+++ b/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml
@@ -27,10 +27,7 @@