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

Remove unnecessary context from mark()/reset() in ItemStream.

Browse Source
This commit is contained in:
dsyer
2008-02-05 10:37:01 +00:00
parent 39210aa3d6
commit 22a1d7c254
25 changed files with 108 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java
View File

@@ -235,7 +235,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
lastCommitRowNumber = currentProcessedRow;
if (!useStatelessSession) {
statefulSession.clear();
@@ -245,7 +245,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
currentProcessedRow = lastCommitRowNumber;
if (lastCommitRowNumber == 0) {
cursor.beforeFirst();

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

@@ -232,7 +232,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
* Mark the current row. Calling reset will cause the result set to be set
* to the current row when mark was called.
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
lastCommittedRow = currentProcessedRow;
skippedRows.clear();
}
@@ -242,7 +242,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource implemen
*
* @throws DataAccessException
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
try {
currentProcessedRow = lastCommittedRow;
if (currentProcessedRow > 0) {

8
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java
View File

@@ -207,11 +207,11 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
}
protected void transactionCommitted() {
mark(null);
mark();
}
protected void transactionRolledBack() {
reset(null);
reset();
}
/**
@@ -237,14 +237,14 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource
/* (non-Javadoc)
* @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
lastCommitIndex = currentIndex;
}
/* (non-Javadoc)
* @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
keysIterator = keys.listIterator(lastCommitIndex);
}

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

@@ -53,7 +53,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
*/
public void open() {
super.open();
mark(null);
mark();
}
/**
@@ -81,7 +81,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
record = readLine();
}
mark(data);
mark();
}
@@ -115,14 +115,14 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
getReader().mark();
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
getReader().reset();
}

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

@@ -118,14 +118,14 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
* Commit the transaction.
*/
protected void transactionCommitted() {
mark(null);
mark();
}
/**
* Rollback the transaction.
*/
protected void transactionRolledBack() {
reset(null);
reset();
}
// This method removes any information in the file before this reset point.
@@ -552,14 +552,14 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
/* (non-Javadoc)
* @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
getOutputState().mark();
}
/* (non-Javadoc)
* @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
getOutputState().checkFileSize();
resetPositionForRestart();
}

5
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/AbstractTransactionalIoSource.java
View File

@@ -17,7 +17,6 @@ package org.springframework.batch.io.support;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.stream.ItemStreamAdapter;
/**
@@ -39,14 +38,14 @@ public abstract class AbstractTransactionalIoSource extends ItemStreamAdapter {
*
* @see TransactionSynchronization#afterCompletion
*/
public abstract void mark(ExecutionAttributes executionAttributes);
public abstract void mark();
/*
* Called when a transaction has been rolled back.
*
* @see TransactionSynchronization#afterCompletion
*/
public abstract void reset(ExecutionAttributes executionAttributes);
public abstract void reset();
/* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#isMarkSupported()

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

@@ -124,7 +124,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
throw new DataAccessResourceFailureException("Unable to get input stream", ioe);
}
initialized = true;
mark(null);
mark();
}
public void setResource(Resource resource) {
@@ -210,7 +210,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
fragmentReader.next();
moveCursorToNextFragment(fragmentReader);
}
mark(null); // reset the history buffer
mark(); // reset the history buffer
}
/**
@@ -266,7 +266,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
lastCommitPointRecordCount = currentRecordCount;
txReader.onCommit();
skipRecords = new ArrayList();
@@ -275,7 +275,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
currentRecordCount = lastCommitPointRecordCount;
txReader.onRollback();
fragmentReader.reset();

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java
View File

@@ -450,7 +450,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
lastCommitPointPosition = getPosition();
lastCommitPointRecordCount = currentRecordCount;
}
@@ -458,7 +458,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
currentRecordCount = lastCommitPointRecordCount;
// close output
close();

36
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java
View File

@@ -25,10 +25,10 @@ package org.springframework.batch.item;
* <p>
* The state that is stored is represented as {@link ExecutionAttributes} which
* enforces a requirement that any restart data can be represented by a
* Properties object. In general, the contract is that {@link ExecutionAttributes}
* that is returned via the {@link #getExecutionAttributes()} method will be given
* back to the {@link #restoreFrom(ExecutionAttributes)} method, exactly as it was
* provided.
* Properties object. In general, the contract is that
* {@link ExecutionAttributes} that is returned via the
* {@link #getExecutionAttributes()} method will be given back to the
* {@link #restoreFrom(ExecutionAttributes)} method, exactly as it was provided.
* </p>
*
* @author Dave Syer
@@ -57,7 +57,14 @@ public interface ItemStream extends ExecutionAttributesProvider {
void close() throws StreamException;
/**
* Clients are expected to check this flag before calling mark or reset.
* Clients are expected to check this flag before calling mark or reset.<br/>
*
* Implementations should also document explicitly, if mark is supported,
* how it will behave in a multi-threaded environment. Generally, if the
* stream is being accessed from multiple threads concurrently, it will have
* to manage that internally, and also reflect only the completed marks
* (independent of the order they happen) when
* {@link ExecutionAttributesProvider#getExecutionAttributes()} is called.
*
* @return true if mark and reset are supported by the {@link ItemStream}
*/
@@ -65,24 +72,25 @@ public interface ItemStream extends ExecutionAttributesProvider {
/**
* Mark the stream so that it can be reset later and the items backed out.
* Implementations may use the information in the provided context to make
* calculations that account for things like multiple open cursors. The
* context should also be updated with any information of this nature that
* might be needed by a reset or by future calls to mark.
* After this method is called the result will be reflected in subsequent
* calls to {@link ExecutionAttributesProvider#getExecutionAttributes()}.<br/>
*
* @param the context which might contain information needed to determine
* what action to take, and into which the current mark information can go.
* In a multi-threaded setting implementations have to ensure that only the
* state from the current thread is saved.
*
* @throws UnsupportedOperationException if the operation is not supported
*/
void mark(ExecutionAttributes executionAttributes);
void mark();
/**
* Reset the stream to the last mark. After a reset the stream state will be
* such that changes (items read or written) since the last call to mark
* with the same context will not be visible after a call to close.
* will not be visible after a call to close.<br/>
*
* In a multi-threaded setting implementations have to ensure that only the
* state from the current thread is reset.
*
* @throws UnsupportedOperationException if the operation is not supported
*/
void reset(ExecutionAttributes executionAttributes);
void reset();
}

8
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java
View File

@@ -119,18 +119,18 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).mark(executionAttributes);
((ItemStream) inputSource).mark();
}
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).reset(executionAttributes);
((ItemStream) inputSource).reset();
}
}
}

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java
View File

@@ -64,14 +64,14 @@ public class ItemStreamAdapter implements ItemStream {
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionAttributes)
*/
public void mark(ExecutionAttributes executionAttributes) {
public void mark() {
throw new UnsupportedOperationException("Mark operation not supported.");
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionAttributes)
*/
public void reset(ExecutionAttributes executionAttributes) {
public void reset() {
throw new UnsupportedOperationException("Reset operation not supported.");
}

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java
View File

@@ -177,7 +177,7 @@ public class SimpleStreamManager implements StreamManager {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
if (stream.isMarkSupported()) {
stream.mark(stream.getExecutionAttributes());
stream.mark();
}
}
});
@@ -186,7 +186,7 @@ public class SimpleStreamManager implements StreamManager {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
if (stream.isMarkSupported()) {
stream.reset(stream.getExecutionAttributes());
stream.reset();
}
}
});