From 6a3cd3ffbe81d4a92b9d201360ec2f71232470d4 Mon Sep 17 00:00:00 2001 From: robokaso Date: Fri, 6 Jun 2008 09:08:18 +0000 Subject: [PATCH] IN PROGRESS - BATCH-653: consistent item buffering in ItemReaders extracted generic buffered reader superclass from HibernateCursorItemReader --- .../AbstractBufferedItemReaderItemStream.java | 171 ++++++++++++++++ .../item/ExecutionContextUserSupport.java | 2 +- .../database/HibernateCursorItemReader.java | 184 ++++++------------ ...tDataSourceItemReaderIntegrationTests.java | 2 +- 4 files changed, 228 insertions(+), 131 deletions(-) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java new file mode 100644 index 000000000..5b565861c --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java @@ -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; + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextUserSupport.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextUserSupport.java index 67e6d3a19..bbc6dd649 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextUserSupport.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContextUserSupport.java @@ -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; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java index 11cecacad..0f22284a5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java @@ -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(); + } + } + + } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java index ed859537a..741ca940f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java @@ -126,7 +126,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends getAsItemStream(reader).open(executionContext); fail(); } - catch (IllegalStateException ex) { + catch (Exception ex) { // expected } }