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

IN PROGRESS - BATCH-653: consistent item buffering in ItemReaders

Browse Source 
extracted generic buffered reader superclass from HibernateCursorItemReader
This commit is contained in:
robokaso
2008-06-06 09:08:18 +00:00
parent e54aafd335
commit 6a3cd3ffbe
4 changed files with 228 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java Normal file
View File

@@ -0,0 +1,171 @@
package org.springframework.batch.item;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import org.springframework.util.Assert;
/**
* Abstract superclass for {@link ItemReader}s which use item buffering to
* support reset/rollback. Supports restart by storing item count in the
* {@link ExecutionContext} (therefore requires item ordering to be preserved
* between runs).
*
* @author Robert Kasanicky
*/
public abstract class AbstractBufferedItemReaderItemStream implements ItemReader, ItemStream {
private static final String ITEM_COUNT = "item.count";
private int currentItemCount = 0;
private int lastMarkedItemCount = 0;
private boolean shouldReadBuffer = false;
private List itemBuffer = new ArrayList();
private ListIterator itemBufferIterator = null;
private int lastMarkedBufferIndex = 0;
private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport();
private boolean saveState = false;
/**
* Read next item from input.
* @return item
* @throws Exception
*/
protected abstract Object doRead() throws Exception;
/**
* Open resources necessary to start reading input.
*/
protected abstract void doOpen() throws Exception;
/**
* Close the resources opened in {@link #doOpen()}.
*/
protected abstract void doClose() throws Exception;
/**
* Move to the given item index. Subclasses should override this method if
* there is a more efficient way of moving to given index than re-reading
* the input using {@link #doRead()}.
*/
protected void jumpToItem(int itemIndex) throws Exception {
for (int i = 0; i < itemIndex; i++) {
doRead();
}
}
public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
currentItemCount++;
if (shouldReadBuffer) {
if (itemBufferIterator.hasNext()) {
return itemBufferIterator.next();
}
else {
// buffer is exhausted, continue reading from file
shouldReadBuffer = false;
itemBufferIterator = null;
}
}
Object item = doRead();
itemBuffer.add(item);
return item;
}
public void mark() throws MarkFailedException {
if (!shouldReadBuffer) {
itemBuffer.clear();
itemBufferIterator = null;
lastMarkedBufferIndex = 0;
}
else {
lastMarkedBufferIndex = itemBufferIterator.nextIndex();
}
lastMarkedItemCount = currentItemCount;
}
public void reset() throws ResetFailedException {
currentItemCount = lastMarkedItemCount;
shouldReadBuffer = true;
itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex);
}
protected int getCurrentItemCount() {
return currentItemCount;
}
protected void setCurrentItemCount(int count) {
this.currentItemCount = count;
}
public void close(ExecutionContext executionContext) throws ItemStreamException {
currentItemCount = 0;
lastMarkedItemCount = 0;
lastMarkedBufferIndex = 0;
itemBufferIterator = null;
shouldReadBuffer = false;
itemBuffer.clear();
try {
doClose();
}
catch (Exception e) {
throw new ItemStreamException("Error while closing item reader", e);
}
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
try {
doOpen();
}
catch (Exception e) {
throw new ItemStreamException("Failed to initialize the reader", e);
}
if (executionContext.containsKey(ecSupport.getKey(ITEM_COUNT))) {
int itemCount = Integer.parseInt(executionContext.getString(ecSupport.getKey(ITEM_COUNT)));
try {
jumpToItem(itemCount);
}
catch (Exception e) {
throw new ItemStreamException("Could not move to stored position on restart", e);
}
currentItemCount = itemCount;
}
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
if (saveState) {
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putString(ecSupport.getKey(ITEM_COUNT), "" + currentItemCount);
}
}
public void setName(String name) {
ecSupport.setName(name);
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
}

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

@@ -41,7 +41,7 @@ public class ExecutionContextUserSupport {
* key that can be safely used to identify data stored in
* {@link ExecutionContext}.
*/
protected String getKey(String s) {
public String getKey(String s) {
Assert.hasLength(name, "ItemStream must have a name assigned.");
return name + "." + s;
}

184
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java
View File

@@ -15,18 +15,13 @@
*/
package org.springframework.batch.item.database;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import org.hibernate.Query;
import org.hibernate.ScrollMode;
import org.hibernate.ScrollableResults;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.hibernate.StatelessSession;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ExecutionContextUserSupport;
import org.springframework.batch.item.AbstractBufferedItemReaderItemStream;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.beans.factory.InitializingBean;
@@ -54,11 +49,9 @@ import org.springframework.util.ClassUtils;
* @author Robert Kasanicky
* @author Dave Syer
*/
public class HibernateCursorItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream,
public class HibernateCursorItemReader extends AbstractBufferedItemReaderItemStream implements ItemStream,
InitializingBean {
private static final String RESTART_DATA_ROW_NUMBER_KEY = "row.number";
private SessionFactory sessionFactory;
private StatelessSession statelessSession;
@@ -71,101 +64,14 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
private boolean useStatelessSession = true;
private int lastCommitRowNumber = 0;
/* Current count of processed records. */
private int currentProcessedRow = 0;
private boolean initialized = false;
private boolean saveState = false;
private boolean shouldReadBuffer = false;
private ListIterator itemBufferIterator = null;
private List itemBuffer = new ArrayList();
private int lastMarkedBufferIndex = 0;
private int fetchSize = 0;
public HibernateCursorItemReader() {
setName(ClassUtils.getShortName(HibernateCursorItemReader.class));
}
public Object read() {
currentProcessedRow++;
if (shouldReadBuffer) {
if (itemBufferIterator.hasNext()) {
return itemBufferIterator.next();
}
else {
// buffer is exhausted, continue reading from file
shouldReadBuffer = false;
itemBufferIterator = null;
}
}
if (cursor.next()) {
Object[] data = cursor.get();
Object item;
if (data.length > 1) {
item = data;
}
else {
item = data[0];
}
itemBuffer.add(item);
return item;
}
return null;
}
/**
* Closes the result set cursor and hibernate session.
*/
public void close(ExecutionContext executionContext) {
initialized = false;
if (cursor != null) {
cursor.close();
}
currentProcessedRow = 0;
if (useStatelessSession) {
if (statelessSession != null) {
statelessSession.close();
}
}
else {
if (statefulSession != null) {
statefulSession.close();
}
}
}
/**
* Creates a forward-only cursor for the query.
*/
public void open(ExecutionContext executionContext) {
Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
cursor = createQuery().setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY);
initialized = true;
if (executionContext.containsKey(getKey(RESTART_DATA_ROW_NUMBER_KEY))) {
currentProcessedRow = Integer.parseInt(executionContext.getString(getKey(RESTART_DATA_ROW_NUMBER_KEY)));
for (int i = 0; i < currentProcessedRow; i++) {
cursor.next();
}
}
}
/**
* Open appropriate type of hibernate session and create the query.
*/
@@ -212,15 +118,6 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
this.useStatelessSession = useStatelessSession;
}
/**
*/
public void update(ExecutionContext executionContext) {
if (saveState) {
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow);
}
}
/**
* Mark is supported as long as this {@link ItemStream} is used in a
* single-threaded environment. The state backing the mark is a single
@@ -231,34 +128,11 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
*/
public void mark() {
if (!shouldReadBuffer) {
itemBuffer.clear();
itemBufferIterator = null;
lastMarkedBufferIndex = 0;
}
else {
lastMarkedBufferIndex = itemBufferIterator.nextIndex();
}
super.mark();
if (!useStatelessSession) {
statefulSession.clear();
}
lastCommitRowNumber = currentProcessedRow;
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.item.stream.ItemStreamAdapter#reset(org.springframework.batch.item.ExecutionContext)
*/
public void reset() {
currentProcessedRow = lastCommitRowNumber;
shouldReadBuffer = true;
itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex);
}
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
/**
@@ -273,4 +147,56 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
protected Object doRead() throws Exception {
if (cursor.next()) {
Object[] data = cursor.get();
Object item;
if (data.length > 1) {
item = data;
}
else {
item = data[0];
}
return item;
}
return null;
}
/**
* Open hibernate session and create a forward-only cursor for the
* {@link #setQueryString(String)}.
*/
protected void doOpen() throws Exception {
Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
cursor = createQuery().setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY);
initialized = true;
}
/**
* Close the cursor and hibernate session.
*/
protected void doClose() throws Exception {
initialized = false;
if (cursor != null) {
cursor.close();
}
if (useStatelessSession) {
if (statelessSession != null) {
statelessSession.close();
}
}
else {
if (statefulSession != null) {
statefulSession.close();
}
}
}
}

2
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java
View File

@@ -126,7 +126,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends
getAsItemStream(reader).open(executionContext);
fail();
}
catch (IllegalStateException ex) {
catch (Exception ex) {
// expected
}
}