Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering)

Browse Source 
http://jira.springframework.org/browse/BATCH-7

More Statistics removal
This commit is contained in:
dsyer
2008-01-31 14:45:39 +00:00
parent 93639e1928
commit c2ccb7ed02
14 changed files with 69 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepContributionTests.java
View File

@@ -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());

2
spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java
View File

@@ -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");

7
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -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}.

4
spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java
View File

@@ -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();

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
View File

@@ -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() {

10
spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java
View File

@@ -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 {
}
}

24
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java
View File

@@ -72,15 +72,11 @@ import org.springframework.util.StringUtils;
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
@@ -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);

18
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java
View File

@@ -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;
}
/**

34
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java
View File

@@ -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;
}

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java
View File

@@ -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();

20
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java
View File

@@ -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));
}
}

12
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java
View File

@@ -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() {

14
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java → spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java
View File

@@ -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;

10
spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java
View File

@@ -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;
}
}