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

BATCH-365: Moved reset and mark to ItemReader interface.

Browse Source
This commit is contained in:
lucasward
2008-02-24 21:14:38 +00:00
parent 58cdd58c79
commit 747a706e86
11 changed files with 126 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java
View File

@@ -16,6 +16,8 @@
package org.springframework.batch.item;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.reader.AbstractItemReader;
/**
@@ -50,5 +52,36 @@ public interface ItemReader {
* @throws Exception if an underlying resource is unavailable.
*/
Object read() throws Exception;
/**
* Mark the stream so that it can be reset later and the items backed out.
* After this method is called the result will be reflected in subsequent
* calls to {@link ExecutionContextProvider#getExecutionContext()}.<br/>
*
* 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
* @throws MarkFailedException if there is a problem with the mark. If a
* mark fails inside a transaction, it would be worrying, but not normally
* fatal.
*/
void mark() throws MarkFailedException;
/**
* 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
* 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
* @throws ResetFailedException if there is a problem with the reset. If a
* reset fails inside a transaction, it would normally be fatal, and would
* leave the stream in an inconsistent state. So while this is an unchecked
* exception, it may be important for a client to catch it explicitly.
*/
void reset() throws ResetFailedException;
}

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

@@ -16,8 +16,6 @@
package org.springframework.batch.item;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
/**
@@ -61,49 +59,4 @@ public interface ItemStream extends ExecutionContextProvider {
* (except open) may throw an exception.
*/
void close() throws StreamException;
/**
* 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 ExecutionContextProvider#getExecutionContext()} is called.
*
* @return true if mark and reset are supported by the {@link ItemStream}
*/
boolean isMarkSupported();
/**
* Mark the stream so that it can be reset later and the items backed out.
* After this method is called the result will be reflected in subsequent
* calls to {@link ExecutionContextProvider#getExecutionContext()}.<br/>
*
* 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
* @throws MarkFailedException if there is a problem with the mark. If a
* mark fails inside a transaction, it would be worrying, but not normally
* fatal.
*/
void mark() throws MarkFailedException;
/**
* 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
* 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
* @throws ResetFailedException if there is a problem with the reset. If a
* reset fails inside a transaction, it would normally be fatal, and would
* leave the stream in an inconsistent state. So while this is an unchecked
* exception, it may be important for a client to catch it explicitly.
*/
void reset() throws ResetFailedException;
}

7
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java
View File

@@ -17,6 +17,8 @@
package org.springframework.batch.item.reader;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
/**
* Base class for {@link ItemReader} implementations.
@@ -25,4 +27,9 @@ import org.springframework.batch.item.ItemReader;
*/
public abstract class AbstractItemReader implements ItemReader {
public void mark() throws MarkFailedException {
}
public void reset() throws ResetFailedException {
}
}

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

@@ -33,10 +33,10 @@ import org.springframework.util.Assert;
*/
public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean, ItemStream {
private ItemReader inputSource;
private ItemReader itemReader;
public void afterPropertiesSet() throws Exception {
Assert.notNull(inputSource, "ItemReader must not be null.");
Assert.notNull(itemReader, "ItemReader must not be null.");
}
/**
@@ -45,7 +45,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemReader#read()
*/
public Object read() throws Exception {
return inputSource.read();
return itemReader.read();
}
/**
@@ -54,8 +54,8 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* {@link ItemStream}.
*/
public ExecutionContext getExecutionContext() {
if (inputSource instanceof ItemStream) {
return ((ItemStream) inputSource).getExecutionContext();
if (itemReader instanceof ItemStream) {
return ((ItemStream) itemReader).getExecutionContext();
}
return new ExecutionContext();
}
@@ -66,8 +66,8 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* {@link ItemStream}.
*/
public void restoreFrom(ExecutionContext data) {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).restoreFrom(data);
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).restoreFrom(data);
}
}
@@ -76,16 +76,16 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @param source
*/
public void setItemReader(ItemReader source) {
this.inputSource = source;
this.itemReader = source;
}
public ItemReader getItemReader() {
return inputSource;
return itemReader;
}
public void skip() {
if (inputSource instanceof Skippable) {
((Skippable) inputSource).skip();
if (itemReader instanceof Skippable) {
((Skippable) itemReader).skip();
}
}
@@ -94,8 +94,8 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemStream#open()
*/
public void open() throws StreamException {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).open();
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).open();
}
}
@@ -104,31 +104,17 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemStream#open()
*/
public void close() throws StreamException {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).close();
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).close();
}
}
/**
* Delegates the call if the delegate is an {@link ItemStream}.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
if (inputSource instanceof ItemStream) {
return ((ItemStream) inputSource).isMarkSupported();
}
return false;
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*/
public void mark() {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).mark();
}
itemReader.mark();
}
/*
@@ -136,8 +122,6 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
if (inputSource instanceof ItemStream) {
((ItemStream) inputSource).reset();
}
itemReader.reset();
}
}

7
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java
View File

@@ -17,6 +17,8 @@
package org.springframework.batch.item.reader;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.batch.support.AbstractMethodInvokingDelegator;
@@ -43,5 +45,10 @@ public class ItemReaderAdapter extends AbstractMethodInvokingDelegator implement
}
public void mark() throws MarkFailedException {
}
public void reset() throws ResetFailedException {
}
}

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

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

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

@@ -29,9 +29,6 @@ import org.springframework.batch.item.exception.StreamException;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.ClassUtils;
/**
@@ -193,28 +190,6 @@ public class SimpleStreamManager implements StreamManager {
*/
public TransactionStatus getTransaction(final Object key) {
TransactionStatus transaction = transactionManager.getTransaction(new DefaultTransactionDefinition());
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
public void afterCompletion(int status) {
if (status == TransactionSynchronization.STATUS_COMMITTED) {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
if (stream.isMarkSupported()) {
stream.mark();
}
}
});
}
else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
iterate(key, new Callback() {
public void execute(ItemStream stream) {
if (stream.isMarkSupported()) {
stream.reset();
}
}
});
}
}
});
return transaction;
}