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 62ce82474..6682e5b9a 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,6 +15,11 @@ */ package org.springframework.batch.item.database; +import java.util.ArrayList; +import java.util.List; +import java.util.ListIterator; + +import org.hibernate.ScrollMode; import org.hibernate.ScrollableResults; import org.hibernate.Session; import org.hibernate.SessionFactory; @@ -25,7 +30,6 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; /** * {@link ItemReader} for reading database records built on top of Hibernate. @@ -71,19 +75,46 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple private boolean saveState = false; + private boolean shouldReadBuffer = false; + + private ListIterator itemBufferIterator = null; + + private List itemBuffer = new ArrayList(); + + private int lastMarkedBufferIndex = 0; + public HibernateCursorItemReader() { - setName(ClassUtils.getShortName(HibernateCursorItemReader.class)); + setName(HibernateCursorItemReader.class.getSimpleName()); } public Object read() { - if (cursor.next()) { - currentProcessedRow++; - Object[] data = cursor.get(); - if (data.length > 1) { - return data; + currentProcessedRow++; + + if (shouldReadBuffer) { + if (itemBufferIterator.hasNext()) { + return itemBufferIterator.next(); } - return data[0]; + 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; } @@ -117,17 +148,19 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple if (useStatelessSession) { statelessSession = sessionFactory.openStatelessSession(); - cursor = statelessSession.createQuery(queryString).scroll(); + cursor = statelessSession.createQuery(queryString).scroll(ScrollMode.FORWARD_ONLY); } else { statefulSession = sessionFactory.openSession(); - cursor = statefulSession.createQuery(queryString).scroll(); + cursor = statefulSession.createQuery(queryString).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))); - cursor.setRowNumber(currentProcessedRow - 1); + for (int i = 0; i < currentProcessedRow; i++) { + cursor.next(); + } } } @@ -181,10 +214,20 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple * @see org.springframework.batch.item.ItemReader#mark() */ public void mark() { - lastCommitRowNumber = currentProcessedRow; + + if (!shouldReadBuffer) { + itemBuffer.clear(); + itemBufferIterator = null; + lastMarkedBufferIndex = 0; + } + else { + lastMarkedBufferIndex = itemBufferIterator.nextIndex(); + } + if (!useStatelessSession) { statefulSession.clear(); } + lastCommitRowNumber = currentProcessedRow; } /* @@ -194,14 +237,8 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple */ public void reset() { currentProcessedRow = lastCommitRowNumber; - if (lastCommitRowNumber == 0) { - cursor.beforeFirst(); - } - else { - // Set the cursor so that next time it is advanced it will - // come back to the committed row. - cursor.setRowNumber(lastCommitRowNumber - 1); - } + shouldReadBuffer = true; + itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex); } public void setSaveState(boolean saveState) { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java index 921c52735..3b0c9714c 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java @@ -63,8 +63,15 @@ public abstract class CommonItemReaderTests extends TestCase { assertEquals(foo2, tested.read()); - assertEquals(foo3, tested.read()); +// TODO handle shortening the commit interval on the fly +// +// tested.mark(); +// +// assertEquals(foo3, tested.read()); +// +// tested.reset(); + assertEquals(foo3, tested.read()); Foo foo4 = (Foo) tested.read(); assertEquals(4, foo4.getValue()); diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/CustomerCredit.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/CustomerCredit.java index 1f0abcb2a..30c08a0bb 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/CustomerCredit.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/CustomerCredit.java @@ -52,8 +52,12 @@ public class CustomerCredit { this.name = name; } - public void increaseCreditBy(BigDecimal sum) { - this.credit = this.credit.add(sum); + public CustomerCredit increaseCreditBy(BigDecimal sum) { + CustomerCredit newCredit = new CustomerCredit(); + newCredit.credit = this.credit.add(sum); + newCredit.name = this.name; + newCredit.id = this.id; + return newCredit; } public boolean equals(Object o) { diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/BatchSqlCustomerCreditIncreaseWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/BatchSqlCustomerCreditIncreaseWriter.java index 48b4e5b1b..f20ccc6ae 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/BatchSqlCustomerCreditIncreaseWriter.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/BatchSqlCustomerCreditIncreaseWriter.java @@ -45,8 +45,7 @@ public class BatchSqlCustomerCreditIncreaseWriter implements ItemWriter, Initial * @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object) */ public void write(Object data) throws Exception { - CustomerCredit customerCredit = (CustomerCredit) data; - customerCredit.increaseCreditBy(FIXED_AMOUNT); + CustomerCredit customerCredit = ((CustomerCredit) data).increaseCreditBy(FIXED_AMOUNT); delegate.write(customerCredit); } diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseWriter.java index beea5a8cc..b10485170 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseWriter.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseWriter.java @@ -29,8 +29,7 @@ public class CustomerCreditIncreaseWriter extends AbstractItemWriter { * @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object) */ public void write(Object data) throws Exception { - CustomerCredit customerCredit = (CustomerCredit) data; - customerCredit.increaseCreditBy(FIXED_AMOUNT); + CustomerCredit customerCredit = ((CustomerCredit) data).increaseCreditBy(FIXED_AMOUNT); customerCreditDao.writeCredit(customerCredit); } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractCustomerCreditIncreaseTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractCustomerCreditIncreaseTests.java index 84107122a..ec0be8f4c 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractCustomerCreditIncreaseTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractCustomerCreditIncreaseTests.java @@ -108,10 +108,13 @@ public abstract class AbstractCustomerCreditIncreaseTests extends AbstractValida public Object mapRow(ResultSet rs, int rowNum) throws SQLException { final BigDecimal creditBeforeUpdate = (BigDecimal) creditsBeforeUpdate.get(rowNum); + System.out.print("BEFORE:" + creditBeforeUpdate); final BigDecimal expectedCredit = creditBeforeUpdate.add(CREDIT_INCREASE); + System.out.print(" EXPECTED:" + expectedCredit); if (expectedCredit.equals(rs.getBigDecimal(CREDIT_COLUMN))) { matches.add(rs.getBigDecimal(ID_COLUMN)); } + System.out.println(" ACTUAL: " + rs.getBigDecimal(CREDIT_COLUMN)); return null; } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseProcessorTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseProcessorTests.java index 2b3255190..badf36306 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseProcessorTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseProcessorTests.java @@ -4,50 +4,41 @@ import java.math.BigDecimal; import junit.framework.TestCase; -import org.easymock.MockControl; import org.springframework.batch.sample.dao.CustomerCreditDao; import org.springframework.batch.sample.domain.CustomerCredit; -import org.springframework.batch.sample.item.writer.CustomerCreditIncreaseWriter; /** * Tests for {@link CustomerCreditIncreaseWriter}. * * @author Robert Kasanicky */ -public class CustomerCreditIncreaseProcessorTests extends TestCase{ +public class CustomerCreditIncreaseProcessorTests extends TestCase { private CustomerCreditIncreaseWriter writer = new CustomerCreditIncreaseWriter(); - - private CustomerCreditDao outputSource; - private MockControl outputSourceControl = MockControl.createStrictControl(CustomerCreditDao.class); - - private CustomerCredit customerCredit = new CustomerCredit(); - - - protected void setUp() throws Exception { - customerCredit.setId(1); - customerCredit.setName("testCustomer"); - - outputSource = (CustomerCreditDao) outputSourceControl.getMock(); - writer.setCustomerCreditDao(outputSource); - } /** * Increases customer's credit by fixed value */ public void testProcess() throws Exception { - BigDecimal oldCredit = new BigDecimal(10.54); + + final BigDecimal oldCredit = new BigDecimal(10.54); + class CustomerDaoStub implements CustomerCreditDao { + + public void writeCredit(CustomerCredit customerCredit) throws Exception { + BigDecimal expectedCredit = oldCredit.add(CustomerCreditIncreaseWriter.FIXED_AMOUNT); + assertTrue(customerCredit.getCredit().compareTo(expectedCredit) == 0); + + } + + } + CustomerCredit customerCredit = new CustomerCredit(); + customerCredit.setId(1); + customerCredit.setName("testCustomer"); + writer.setCustomerCreditDao(new CustomerDaoStub()); + customerCredit.setCredit(oldCredit); - - outputSource.writeCredit(customerCredit); - outputSourceControl.setVoidCallable(); - outputSourceControl.replay(); - + writer.write(customerCredit); - - BigDecimal newCredit = customerCredit.getCredit(); - BigDecimal expectedCredit = oldCredit.add(CustomerCreditIncreaseWriter.FIXED_AMOUNT); - assertTrue(newCredit.compareTo(expectedCredit) == 0); - outputSourceControl.verify(); + } }