diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepContributionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepContributionTests.java index ebe2be306..4cb7cdbe8 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepContributionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepContributionTests.java @@ -45,7 +45,7 @@ public class StepContributionTests extends TestCase { * Test method for * {@link org.springframework.batch.core.domain.StepContribution#setStreamContext(StreamContext)}. */ - public void testSetStatistics() { + public void testSetStreamContext() { assertEquals(null, contribution.getStreamContext()); contribution.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties("foo=bar"))); assertEquals(1, contribution.getStreamContext().getProperties().size()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java index cef3c10d0..699c9a3bc 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java @@ -198,7 +198,7 @@ public class StepExecutionTests extends TestCase { execution.toString().indexOf("rollback") >= 0); } - public void testStatistics() throws Exception { + public void testStreamContext() throws Exception { assertNotNull(execution.getStreamContext()); StreamContext context = new StreamContext(); context.putString("foo", "bar"); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index f5d292c8d..08232966f 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -33,6 +33,7 @@ import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.io.Skippable; import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.item.stream.SimpleStreamManager; @@ -105,9 +106,9 @@ public class SimpleStepExecutor { /** * Public setter for the {@link StreamManager}. This will be used to create * the {@link StepContext}, and hence any component that is a - * {@link StatisticsProvider} and in step scope will be registered with the - * service. The {@link StepContext} is then a source of aggregate statistics - * for the step. + * {@link ItemStream} and in step scope will be registered with the service. The + * {@link StepContext} is then a source of aggregate statistics for the + * step. * * @param streamManager the {@link StreamManager} to set. Default is a * {@link SimpleStreamManager}. diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java index 9c5fc2635..9b74926b6 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java @@ -134,14 +134,14 @@ public class SimpleStepContextTests extends TestCase { assertTrue(list.contains("spam")); } - public void testStatisticsWithNotNullService() throws Exception { + public void testStreamContextWithNotNullService() throws Exception { Map map = new HashMap(); context = new SimpleStepContext(null, null, new StubStreamManager(map)); assertEquals(1, context.getStreamContext().getProperties().size()); assertEquals("bar", context.getStreamContext().getProperties().getProperty("foo")); } - public void testStatisticsServiceRegistration() throws Exception { + public void testStreamManagerRegistration() throws Exception { Map map = new HashMap(); context = new SimpleStepContext(null, null, new StubStreamManager(map)); ItemStreamAdapter provider = new ItemStreamAdapter(); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index b995b297c..a806ddd05 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -409,7 +409,7 @@ public class SimpleStepExecutorTests extends TestCase { JobExecution jobExecution = new JobExecution(jobInstance); StepExecution stepExecution = new StepExecution(step, jobExecution); - assertEquals(null, stepExecution.getStreamContext().getString("foo")); + assertEquals(false, stepExecution.getStreamContext().containsKey("foo")); final Map map = new HashMap(); stepExecutor.setStreamManager(new SimpleStreamManager() { diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java index 2712ffbfe..672ab668f 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java @@ -19,7 +19,6 @@ package org.springframework.batch.execution.tasklet; import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.Properties; import junit.framework.TestCase; @@ -34,7 +33,6 @@ import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.context.RepeatContextSupport; import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; import org.springframework.batch.retry.policy.SimpleRetryPolicy; -import org.springframework.batch.support.PropertiesConverter; /** * @author Dave Syer @@ -292,10 +290,6 @@ public class ItemOrientedTaskletTests extends TestCase { list.add("provider"); } - public Properties getStatistics() { - return PropertiesConverter.stringToProperties("foo=bar"); - } - public void close() throws StreamException { } } @@ -320,10 +314,6 @@ public class ItemOrientedTaskletTests extends TestCase { list.add("writer"); } - public Properties getStatistics() { - return PropertiesConverter.stringToProperties(props); - } - public void close() throws Exception { } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java index fa6d456e1..327d49535 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java @@ -72,15 +72,11 @@ import org.springframework.util.StringUtils; *

* *

- * Restart: This implementation contains basic, simple restart. The current row - * is returned as restart data, and when restored from that same data, the - * cursor is opened and the current row set to the value within the restart - * data. - *

- * - *

- * Statistics: There are two statistics returned by this input source: the - * current line being processed and the number of lines that have been skipped. + * {@link StreamContext}: The current row is returned as restart data, and when + * restored from that same data, the cursor is opened and the current row set to + * the value within the restart data. There are also two statistics returned by + * this input source: the current line being processed and the number of lines + * that have been skipped. *

* *

@@ -216,7 +212,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen } } - Object mappedResult = mapper.mapRow(rs, (int)currentProcessedRow); + Object mappedResult = mapper.mapRow(rs, (int) currentProcessedRow); verifyCursorPosition(currentProcessedRow); @@ -251,7 +247,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen try { currentProcessedRow = lastCommittedRow; if (currentProcessedRow > 0) { - rs.absolute((int)currentProcessedRow); + rs.absolute((int) currentProcessedRow); } else { rs.beforeFirst(); @@ -405,8 +401,8 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen return context; } - - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.StreamContext) */ public void restoreFrom(StreamContext data) { @@ -424,7 +420,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen try { this.currentProcessedRow = data.getLong(CURRENT_PROCESSED_ROW); - rs.absolute((int)currentProcessedRow); + rs.absolute((int) currentProcessedRow); } catch (SQLException se) { throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java index 531b9cd13..31e439ccb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java @@ -17,7 +17,6 @@ package org.springframework.batch.io.file; import java.util.HashSet; -import java.util.Properties; import java.util.Set; import org.apache.commons.logging.Log; @@ -27,7 +26,6 @@ import org.springframework.batch.io.file.separator.LineReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; -import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationAdapter; @@ -54,8 +52,6 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen private TransactionSynchronization transactionSynchronization = new ResourceLineReaderTransactionSynchronization(); - private Properties statistics = new Properties(); - /** * Initialize the input source. */ @@ -104,19 +100,13 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen * case of restart. */ public StreamContext getStreamContext() { - return new GenericStreamContext(getStatistics()); - } - - /** - * @return statistics for input template - * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() - */ - public Properties getStatistics() { if (reader==null) { throw new StreamException("ItemStream not open or already closed."); } - statistics.setProperty(READ_STATISTICS_NAME, String.valueOf(reader.getCurrentLineCount())); - return statistics; + StreamContext streamContext = new StreamContext(); + streamContext.putLong(READ_STATISTICS_NAME, reader.getCurrentLineCount()); + streamContext.putLong(SKIPPED_STATISTICS_NAME, skippedLines.size()); + return streamContext; } /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java index d5ae418c2..52d6ec0fe 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java @@ -34,6 +34,7 @@ import org.springframework.batch.io.support.AbstractTransactionalIoSource; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.item.writer.ItemTransformer; import org.springframework.beans.factory.DisposableBean; @@ -71,11 +72,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements private Resource resource; - private Properties statistics = new Properties(); - private StreamContext streamContext = new GenericStreamContext(new Properties()); - private OutputState state = new OutputState(); + private OutputState state = null; private ItemTransformer transformer = new ItemTransformer() { public Object transform(Object input) { @@ -201,7 +200,10 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements * @see ResourceLifecycle#close() */ public void close() { - getOutputState().close(); + if (state!=null) { + getOutputState().close(); + state = null; + } } /** @@ -240,27 +242,20 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements * @see ResourceLifecycle#open() */ public void open() { + getOutputState(); super.registerSynchronization(); } - /** - * @see StatisticsProvider - */ - public Properties getStatistics() { - final OutputState os = getOutputState(); - - statistics.setProperty(WRITTEN_STATISTICS_NAME, String.valueOf(os.linesWritten)); - statistics.setProperty(RESTART_COUNT_STATISTICS_NAME, String.valueOf(os.restartCount)); - return statistics; - } - /** * @see ItemStream#getStreamContext() */ public StreamContext getStreamContext() { - final OutputState os = getOutputState(); - - streamContext.putString(RESTART_DATA_NAME, String.valueOf(os.position())); + if (state == null) { + throw new StreamException("ItemStream not open or already closed."); + } + streamContext.putLong(RESTART_DATA_NAME, state.position()); + streamContext.putLong(WRITTEN_STATISTICS_NAME, state.linesWritten); + streamContext.putLong(RESTART_COUNT_STATISTICS_NAME, state.restartCount); return streamContext; } @@ -277,6 +272,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements // Returns object representing state. private OutputState getOutputState() { + if (state==null) { + state = new OutputState(); + } return (OutputState) state; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java index 4c0703697..b90ef1e98 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java @@ -99,6 +99,9 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade public void close() { initialized = false; + if (fragmentReader==null && inputStream==null) { + return; + } try { fragmentReader.close(); inputStream.close(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java index 4f8e1fe0e..aea1eadae 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java @@ -17,14 +17,12 @@ package org.springframework.batch.io.file; import java.io.IOException; -import java.util.Properties; import junit.framework.TestCase; -import org.springframework.batch.io.file.DefaultFlatFileItemReader; import org.springframework.batch.io.file.mapping.DefaultFieldSet; -import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.mapping.FieldSet; +import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.transform.LineTokenizer; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; @@ -163,16 +161,16 @@ public class DefaultFlatFileItemReaderTests extends TestCase { inputSource.skip(); inputSource.read(); - Properties statistics = inputSource.getStatistics(); - String skipped = statistics.getProperty(DefaultFlatFileItemReader.SKIPPED_STATISTICS_NAME); - String read = statistics.getProperty(DefaultFlatFileItemReader.READ_STATISTICS_NAME); + StreamContext statistics = inputSource.getStreamContext(); + long skipped = statistics.getLong(DefaultFlatFileItemReader.SKIPPED_STATISTICS_NAME); + long read = statistics.getLong(DefaultFlatFileItemReader.READ_STATISTICS_NAME); // call unknown, which has no influence and therefore statistics should // be the same inputSource.getTransactionSynchronization().afterCompletion(TransactionSynchronization.STATUS_UNKNOWN); - statistics = inputSource.getStatistics(); - assertEquals(skipped, statistics.getProperty(DefaultFlatFileItemReader.SKIPPED_STATISTICS_NAME)); - assertEquals(read, statistics.getProperty(DefaultFlatFileItemReader.READ_STATISTICS_NAME)); + statistics = inputSource.getStreamContext();; + assertEquals(skipped, statistics.getLong(DefaultFlatFileItemReader.SKIPPED_STATISTICS_NAME)); + assertEquals(read, statistics.getLong(DefaultFlatFileItemReader.READ_STATISTICS_NAME)); } public void testRestartFromNullData() throws Exception { @@ -225,8 +223,8 @@ public class DefaultFlatFileItemReaderTests extends TestCase { assertEquals("[testLine5]", inputSource.read().toString()); assertEquals("[testLine6]", inputSource.read().toString()); - Properties statistics = inputSource.getStatistics(); - assertEquals("6", statistics.getProperty(DefaultFlatFileItemReader.READ_STATISTICS_NAME)); + StreamContext statistics = inputSource.getStreamContext(); + assertEquals(6, statistics.getLong(DefaultFlatFileItemReader.READ_STATISTICS_NAME)); } } 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 616736c32..8c35a2730 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 @@ -21,7 +21,6 @@ import java.io.File; import java.io.FileReader; import java.io.IOException; import java.util.Collections; -import java.util.Properties; import junit.framework.TestCase; @@ -345,6 +344,8 @@ public class FlatFileItemWriterTests extends TestCase { inputSource.write("testLine7"); inputSource.write("testLine8"); + // get statistics + StreamContext statistics = inputSource.getStreamContext(); // close template inputSource.close(); @@ -353,10 +354,8 @@ public class FlatFileItemWriterTests extends TestCase { assertEquals("testLine" + i, readLine()); } - // get statistics - Properties statistics = inputSource.getStatistics(); // 3 lines were written to the file after restart - assertEquals("3", statistics.getProperty(FlatFileItemWriter.WRITTEN_STATISTICS_NAME)); + assertEquals(3, statistics.getLong(FlatFileItemWriter.WRITTEN_STATISTICS_NAME)); } @@ -373,10 +372,11 @@ public class FlatFileItemWriterTests extends TestCase { public void testDefaultStreamContext() throws Exception { inputSource = new FlatFileItemWriter(); + inputSource.open(); StreamContext streamContext = inputSource.getStreamContext(); assertNotNull(streamContext); - assertEquals(1, streamContext.getProperties().size()); - assertEquals("0", streamContext.getProperties().getProperty(FlatFileItemWriter.RESTART_DATA_NAME)); + assertEquals(3, streamContext.getProperties().size()); + assertEquals(0, streamContext.getLong(FlatFileItemWriter.RESTART_DATA_NAME)); } private void commit() { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java similarity index 93% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java index aac2f64c1..a1632488b 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java @@ -29,7 +29,7 @@ import org.springframework.transaction.support.TransactionSynchronization; * * @author Robert Kasanicky */ -public class StaxEventReaderItemReaderTests extends TestCase { +public class StaxEventItemReaderTests extends TestCase { // object under test private StaxEventItemReader source; @@ -216,9 +216,13 @@ public class StaxEventReaderItemReaderTests extends TestCase { assertEquals(NUMBER_OF_RECORDS, extractRecordCountFrom(source.getStatistics())); } - public void testClose() throws Exception{ + public void testCloseWithoutOpen() throws Exception{ + source.close(); + // No error! + } - MockStaxEventReaderItemReader newSource = new MockStaxEventReaderItemReader(); + public void testClose() throws Exception{ + MockStaxEventItemReader newSource = new MockStaxEventItemReader(); Resource resource = new ByteArrayResource(xml.getBytes()); newSource.setResource(resource); @@ -229,7 +233,7 @@ public class StaxEventReaderItemReaderTests extends TestCase { assertNotNull(item); assertTrue(newSource.isOpenCalled()); - newSource.destroy(); + newSource.destroy(); // includes close() newSource.setOpenCalled(false); //calling read again should require re-initialization because of close item = newSource.read(); @@ -375,7 +379,7 @@ public class StaxEventReaderItemReaderTests extends TestCase { } - private static class MockStaxEventReaderItemReader extends StaxEventItemReader { + private static class MockStaxEventItemReader extends StaxEventItemReader { private boolean openCalled = false; diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java index 51caf3f3c..4f43b1a78 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java @@ -16,14 +16,11 @@ package org.springframework.batch.sample.tasklet; -import java.util.Properties; - import org.springframework.batch.core.tasklet.Tasklet; import org.springframework.batch.execution.tasklet.ItemOrientedTasklet; import org.springframework.batch.io.file.DefaultFlatFileItemReader; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamContextProvider; -import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.sample.dao.TradeDao; import org.springframework.batch.sample.domain.Trade; @@ -90,10 +87,9 @@ public class SimpleTradeTasklet implements Tasklet, StreamContextProvider { * @see org.springframework.batch.item.StreamContextProvider#getStreamContext() */ public StreamContext getStreamContext() { - Properties statistics = new Properties(); - statistics.setProperty("trade.count", String.valueOf(tradeCount)); - statistics.putAll(inputSource.getStatistics()); - return new GenericStreamContext(statistics); + StreamContext statistics = new StreamContext(); + statistics.putLong("trade.count", tradeCount); + return statistics; } }