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

Incomplete - task 84: Fix samples

Browse Source 
Fix column mapping in jdbc cursor - order is important.
This commit is contained in:
dsyer
2008-02-26 10:41:30 +00:00
parent d20c76b0a7
commit 7f0be2826e
35 changed files with 199 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -187,7 +187,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
/**
*/
public void beforeSave() {
public void update() {
executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow);
String skipped = skippedRows.toString();
executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));
@@ -210,15 +210,7 @@ public class HibernateCursorItemReader extends AbstractItemStreamItemReader impl
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#mark(org.springframework.batch.item.ExecutionContext)
* @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
lastCommitRowNumber = currentProcessedRow;

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

@@ -385,7 +385,7 @@ public class JdbcCursorItemReader implements KeyedItemReader, InitializingBean,
* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#getExecutionContext()
*/
public void beforeSave() {
public void update() {
if(saveState){
String skipped = skippedRows.toString();
executionContext.putString(addName(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1));

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

@@ -146,7 +146,7 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
this.executionContext = executionContext;
}
public void beforeSave() {
public void update() {
if(saveState){
if(getCurrentKey() != null){
keyGenerator.saveState(getCurrentKey(), executionContext);
@@ -190,16 +190,7 @@ public class DrivingQueryItemReader implements KeyedItemReader, InitializingBean
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionContext)
* @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
lastCommitIndex = currentIndex;

11
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapExecutionContextRowMapper.java
View File

@@ -38,10 +38,11 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
public PreparedStatementSetter createSetter(ExecutionContext executionContext) {
List columns = new ArrayList();
for (Iterator iterator = executionContext.entrySet().iterator(); iterator.hasNext();) {
Entry entry = (Entry) iterator.next();
Object column = entry.getValue();
int count=0;
while(executionContext.containsKey(KEY_PREFIX+count)) {
Object column = executionContext.get(KEY_PREFIX+count);
columns.add(column);
count++;
}
return new ArgPreparedStatementSetter(columns.toArray());
@@ -50,9 +51,11 @@ public class ColumnMapExecutionContextRowMapper extends ColumnMapRowMapper imple
public void mapKeys(Object key, ExecutionContext executionContext) {
Assert.isInstanceOf(Map.class, key, "Input to create ExecutionContext must be of type Map.");
Map keys = (Map) key;
int count = 0;
for (Iterator it = keys.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry)it.next();
executionContext.put(entry.getKey().toString(), entry.getValue());
executionContext.put(KEY_PREFIX+count, entry.getValue());
count++;
}
}

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java
View File

@@ -90,7 +90,7 @@ public class MultipleColumnJdbcKeyGenerator implements
* @see org.springframework.batch.io.driving.KeyGenerator#getKeyAsExecutionContext(java.lang.Object)
*/
public void saveState(Object key, ExecutionContext executionContext) {
Assert.state(keyMapper != null, "Kye mapper must not be null.");
Assert.state(keyMapper != null, "Key mapper must not be null.");
keyMapper.mapKeys(key, executionContext);
}

12
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java
View File

@@ -202,7 +202,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* the current Line Count which can be used to reinitialise the batch job in
* case of restart.
*/
public void beforeSave() {
public void update() {
if (reader == null) {
throw new StreamException("ItemStream not open or already closed.");
}
@@ -216,15 +216,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
* @see org.springframework.batch.item.ItemReader#mark()
*/
public void mark() {
getReader().mark();

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

@@ -190,9 +190,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
/**
* @see ItemStream#beforeSave()
* @see ItemStream#update()
*/
public void beforeSave() {
public void update() {
if (state == null) {
throw new StreamException("ItemStream not open or already closed.");
}
@@ -471,34 +471,6 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements
}
/**
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/*
* 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)
* @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() throws ResetFailedException {
}
public void clear() throws Exception {
try {
getOutputState().reset();

15
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java
View File

@@ -195,22 +195,15 @@ public class ResourceLineReader extends ItemStreamSupport implements LineReader,
}
/**
* Mark the state for return later with reset. Uses the read-ahead limit
* from an underlying {@link BufferedReader}, which means that there is a
* limit to how much data can be recovered if the mark needs to be reset.<br/>
*
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/**
* Mark the state for return later with reset. Uses the read-ahead limit
* from an underlying {@link BufferedReader}, which means that there is a
* limit to how much data can be recovered if the mark needs to be reset.
*
* @see #reset()
*
* @throws MarkFailedException if the mark could not be set.

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

@@ -32,26 +32,5 @@ import org.springframework.batch.item.stream.ItemStreamSupport;
* @since 1.0
*/
public abstract class AbstractTransactionalIoSource extends ItemStreamSupport {
/*
* Called when a transaction has been committed.
*
* @see TransactionSynchronization#afterCompletion
*/
public abstract void mark();
/*
* Called when a transaction has been rolled back.
*
* @see TransactionSynchronization#afterCompletion
*/
public abstract void reset();
/* (non-Javadoc)
* @see org.springframework.batch.item.stream.ItemStreamAdapter#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
}

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

@@ -193,9 +193,9 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
}
/**
* @see ItemStream#beforeSave()
* @see ItemStream#update()
*/
public void beforeSave() {
public void update() {
executionContext.putLong(READ_COUNT_STATISTICS_NAME, currentRecordCount);
}
@@ -254,16 +254,8 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*
* @see org.springframework.batch.item.reader.AbstractItemReader#mark()
*/
public void mark() {
lastCommitPointRecordCount = currentRecordCount;

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

@@ -363,9 +363,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
/**
* Get the restart data.
* @see org.springframework.batch.item.ItemStream#beforeSave()
* @see org.springframework.batch.item.ItemStream#update()
*/
public void beforeSave() {
public void update() {
if (!initialized) {
throw new StreamException("ItemStream is not open, or may have been closed. Cannot access context.");
}
@@ -421,38 +421,11 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing
}
}
/**
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
* counter, keeping track of the current position, so multiple threads
* cannot be accommodated.
*
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/* TODO remove once ItemStream interface is modified.
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext)
*/
public void mark() {
}
public void flush() throws Exception {
lastCommitPointPosition = getPosition();
lastCommitPointRecordCount = currentRecordCount;
}
/* TODO remove once ItemStream interface is modified.
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
}
public void clear() throws Exception {
currentRecordCount = lastCommitPointRecordCount;

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

@@ -52,22 +52,19 @@ 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#beforeSave()}.<br/>
* Mark the stream so that it can be reset later and the items backed out.<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
@@ -76,7 +73,6 @@ public interface ItemReader {
* 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

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

@@ -43,7 +43,7 @@ public interface ItemStream {
* has not been put in the context, it should be added
* here.
*/
void beforeSave();
void update();
/**
* If any resources are needed for the stream to operate they need to be

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

@@ -50,13 +50,13 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl
}
/**
* @see ItemStream#beforeSave()
* @see ItemStream#update()
* @throws IllegalStateException if the parent template is not itself
* {@link ItemStream}.
*/
public void beforeSave() {
public void update() {
if (itemReader instanceof ItemStream) {
((ItemStream) itemReader).beforeSave();
((ItemStream) itemReader).update();
}
}

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

@@ -43,9 +43,9 @@ public class ItemStreamSupport implements ItemStream {
/**
* Return empty {@link ExecutionContext}.
* @see org.springframework.batch.item.ExecutionContextProvider#beforeSave()
* @see org.springframework.batch.item.ExecutionContextProvider#update()
*/
public void beforeSave() {
public void update() {
}
}

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

@@ -68,11 +68,11 @@ public class SimpleStreamManager implements StreamManager {
*
* @see org.springframework.batch.item.stream.StreamManager#getExecutionContext(java.lang.Object)
*/
public void beforeSave() {
synchronized(streams){
for(Iterator it = streams.iterator(); it.hasNext();){
ItemStream itemStream = (ItemStream)it.next();
itemStream.beforeSave();
public void update() {
synchronized (streams) {
for (Iterator it = streams.iterator(); it.hasNext();) {
ItemStream itemStream = (ItemStream) it.next();
itemStream.update();
}
}
}
@@ -86,7 +86,9 @@ public class SimpleStreamManager implements StreamManager {
*/
public void register(ItemStream stream) {
synchronized (streams) {
streams.add(stream);
if (!streams.contains(stream)) {
streams.add(stream);
}
}
}
@@ -95,9 +97,9 @@ public class SimpleStreamManager implements StreamManager {
* @throws StreamException
*/
public void close() throws StreamException {
synchronized(streams){
for(Iterator it = streams.iterator(); it.hasNext();){
ItemStream itemStream = (ItemStream)it.next();
synchronized (streams) {
for (Iterator it = streams.iterator(); it.hasNext();) {
ItemStream itemStream = (ItemStream) it.next();
itemStream.close();
}
}
@@ -108,9 +110,9 @@ public class SimpleStreamManager implements StreamManager {
* @throws StreamException
*/
public void open(ExecutionContext executionContext) throws StreamException {
synchronized(streams){
for(Iterator it = streams.iterator(); it.hasNext();){
ItemStream itemStream = (ItemStream)it.next();
synchronized (streams) {
for (Iterator it = streams.iterator(); it.hasNext();) {
ItemStream itemStream = (ItemStream) it.next();
itemStream.open(executionContext);
}
}

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

@@ -49,7 +49,7 @@ public interface StreamManager {
* @return {@link ExecutionContext} aggregating the contexts of all providers
* registered under this key, or empty otherwise.
*/
void beforeSave();
void update();
/**
* If any resources are needed for the stream to operate they need to be

6
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java
View File

@@ -80,10 +80,10 @@ public class DelegatingItemWriter implements ItemWriter, ItemStream, Initializin
}
/**
* @see org.springframework.batch.item.ExecutionContextProvider#beforeSave()
* @see org.springframework.batch.item.ExecutionContextProvider#update()
*/
public void beforeSave() {
stream.beforeSave();
public void update() {
stream.update();
}
/**