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

RESOLVED - issue BATCH-1493: Step listeners detected and invoked twice

Browse Source 
RESOLVED - issue BATCH-1494: FetchSize not accessible in HibernateCursorItemReader
This commit is contained in:
dsyer
2010-01-28 11:35:41 +00:00
parent 7c19e7aef3
commit 5d5ce07c8a
7 changed files with 108 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -15,6 +15,9 @@
*/
package org.springframework.batch.core.step.item;
import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ChunkListener;
@@ -600,22 +603,29 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
*/
private void registerItemListeners(SimpleChunkProvider<T> chunkProvider, SimpleChunkProcessor<T, S> chunkProcessor) {
StepListener[] listeners = getListeners();
// explicitly set item listeners
chunkProvider.setListeners(BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(getListeners(),
chunkProvider.setListeners(BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(listeners,
ItemReadListener.class));
chunkProvider.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
chunkProvider.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(listeners,
SkipListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(getListeners(),
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(listeners,
ItemProcessListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(getListeners(),
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(listeners,
ItemWriteListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
chunkProcessor.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(listeners,
SkipListener.class));
List<StepListener> listofListeners = Arrays.asList(listeners);
// auto-register reader, processor and writer
for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) {
if (listofListeners.contains(itemHandler)) {
continue;
}
if (StepListenerFactoryBean.isListener(itemHandler)) {
StepListener listener = StepListenerFactoryBean.getListener(itemHandler);
if (listener instanceof SkipListener<?,?>) {

40
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
View File

@@ -385,6 +385,46 @@ public class SimpleStepFactoryBeanTests {
}
}
@Test
public void testAutoRegisterItemListenersNoDoubleCounting() throws Exception {
SimpleStepFactoryBean<String, String> factory = getStepFactory(new String[] { "foo", "bar", "spam" });
final List<String> listenerCalls = new ArrayList<String>();
class TestItemListenerWriter implements ItemWriter<String>, ItemWriteListener<String> {
public void write(List<? extends String> items) throws Exception {
}
public void afterWrite(List<? extends String> items) {
listenerCalls.add("write");
}
public void beforeWrite(List<? extends String> items) {
}
public void onWriteError(Exception exception, List<? extends String> items) {
}
}
TestItemListenerWriter itemWriter = new TestItemListenerWriter();
factory.setListeners(new StepListener[] {itemWriter});
factory.setItemWriter(itemWriter);
Step step = (Step) factory.getObject();
job.setSteps(Collections.singletonList(step));
JobExecution jobExecution = repository.createJobExecution(job.getName(), new JobParameters());
job.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
assertEquals("[write, write, write]", listenerCalls.toString());
}
private SimpleStepFactoryBean<String, String> getStepFactory(String... args) throws Exception {
SimpleStepFactoryBean<String, String> factory = new SimpleStepFactoryBean<String, String>();

5
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java
View File

@@ -391,9 +391,12 @@ public abstract class AbstractCursorItemReader<T> extends AbstractItemCountingIt
/**
* Execute the statement to open the cursor.
*/
protected void doOpen() throws Exception {
@Override
protected final void doOpen() throws Exception {
Assert.state(!initialized, "Stream is already initialized. Close before re-opening.");
Assert.isNull(rs, "ResultSet still open! Close before re-opening.");
initializeConnection();
openCursor(con);
initialized = true;

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

@@ -68,7 +68,12 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
private boolean initialized = false;
private int fetchSize;
private Map<String, Object> parameterValues;
public void afterPropertiesSet() throws Exception {
Assert.state(fetchSize >= 0, "fetchSize must not be negative");
helper.afterPropertiesSet();
}
@@ -78,7 +83,7 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
* @param parameterValues the parameter values to set
*/
public void setParameterValues(Map<String, Object> parameterValues) {
helper.setParameterValues(parameterValues);
this.parameterValues = parameterValues;
}
/**
@@ -93,6 +98,16 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
helper.setQueryName(queryName);
}
/**
* Fetch size used internally by Hibernate to limit amount of data fetched
* from database per round trip.
*
* @param fetchSize the fetch size to pass down to Hibernate
*/
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
/**
* A query provider. Either this or the {{@link #setQueryString(String)
* query string} or the {{@link #setQueryName(String) query name} should be
@@ -165,7 +180,7 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
*/
protected void doOpen() throws Exception {
Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
cursor = helper.getForwardOnlyCursor();
cursor = helper.getForwardOnlyCursor(fetchSize, parameterValues);
initialized = true;
}
@@ -193,7 +208,8 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
*/
@Override
protected void jumpToItem(int itemIndex) throws Exception {
helper.jumpToItem(cursor, itemIndex);
int flushSize = Math.max(fetchSize, 100);
helper.jumpToItem(cursor, itemIndex, flushSize);
}
/**

31
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java
View File

@@ -46,14 +46,10 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
private String queryName = "";
private Map<String, Object> parameterValues;
private HibernateQueryProvider queryProvider;
private boolean useStatelessSession = true;
private int fetchSize = 0;
private StatelessSession statelessSession;
private Session statefulSession;
@@ -92,15 +88,6 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
this.useStatelessSession = useStatelessSession;
}
/**
* The parameter values to apply to a query (map of name:value).
*
* @param parameterValues the parameter values to set
*/
public void setParameterValues(Map<String, Object> parameterValues) {
this.parameterValues = parameterValues;
}
/**
* @param sessionFactory hibernate session factory
*/
@@ -111,7 +98,6 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
public void afterPropertiesSet() throws Exception {
Assert.state(sessionFactory != null, "A SessionFactory must be provided");
Assert.state(fetchSize >= 0, "fetchSize must not be negative");
if (queryProvider == null) {
Assert.notNull(sessionFactory, "session factory must be set");
@@ -129,9 +115,12 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
/**
* Get a cursor over all of the results, with the forward-only flag set.
*
* @param fetchSize the fetch size to use retrieving the results
* @param parameterValues the parameter values to use (or null if none).
*
* @return a forward-only {@link ScrollableResults}
*/
public ScrollableResults getForwardOnlyCursor() {
public ScrollableResults getForwardOnlyCursor(int fetchSize, Map<String, Object> parameterValues) {
Query query = createQuery();
if (parameterValues != null) {
query.setProperties(parameterValues);
@@ -142,7 +131,7 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
/**
* Open appropriate type of hibernate session and create the query.
*/
private Query createQuery() {
public Query createQuery() {
if (useStatelessSession) {
statelessSession = sessionFactory.openStatelessSession();
@@ -183,11 +172,10 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
*
* @param cursor the results to scroll over
*/
public void jumpToItem(ScrollableResults cursor, int itemIndex) {
int flushSize = Math.max(fetchSize, 100);
public void jumpToItem(ScrollableResults cursor, int itemIndex, int flushInterval) {
for (int i = 0; i < itemIndex; i++) {
cursor.next();
if (i % flushSize == 0 && !useStatelessSession) {
if (i % flushInterval == 0 && !useStatelessSession) {
statefulSession.clear(); // Clears in-memory cache
}
}
@@ -211,9 +199,12 @@ public class HibernateItemReaderHelper<T> implements InitializingBean {
*
* @param page the page to read (starting at 0)
* @param pageSize the size of the page or maximum number of items to read
* @param fetchSize the fetch size to use
* @param parameterValues the parameter values to use (if any, otherwise
* null)
* @return a collection of items
*/
public Collection<? extends T> readPage(int page, int pageSize) {
public Collection<? extends T> readPage(int page, int pageSize, int fetchSize, Map<String, Object> parameterValues) {
clear();

20
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java
View File

@@ -26,6 +26,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.database.orm.HibernateQueryProvider;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
@@ -61,6 +62,10 @@ public class HibernatePagingItemReader<T> extends AbstractPagingItemReader<T> im
private HibernateItemReaderHelper<T> helper = new HibernateItemReaderHelper<T>();
private Map<String, Object> parameterValues;
private int fetchSize;
public HibernatePagingItemReader() {
setName(ClassUtils.getShortName(HibernatePagingItemReader.class));
}
@@ -71,7 +76,7 @@ public class HibernatePagingItemReader<T> extends AbstractPagingItemReader<T> im
* @param parameterValues the parameter values to set
*/
public void setParameterValues(Map<String, Object> parameterValues) {
helper.setParameterValues(parameterValues);
this.parameterValues = parameterValues;
}
/**
@@ -86,6 +91,16 @@ public class HibernatePagingItemReader<T> extends AbstractPagingItemReader<T> im
helper.setQueryName(queryName);
}
/**
* Fetch size used internally by Hibernate to limit amount of data fetched
* from database per round trip.
*
* @param fetchSize the fetch size to pass down to Hibernate
*/
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
/**
* A query provider. Either this or the {{@link #setQueryString(String)
* query string} or the {{@link #setQueryName(String) query name} should be
@@ -130,6 +145,7 @@ public class HibernatePagingItemReader<T> extends AbstractPagingItemReader<T> im
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
Assert.state(fetchSize >= 0, "fetchSize must not be negative");
helper.afterPropertiesSet();
}
@@ -148,7 +164,7 @@ public class HibernatePagingItemReader<T> extends AbstractPagingItemReader<T> im
results.clear();
}
results.addAll(helper.readPage(getPage(), getPageSize()));
results.addAll(helper.readPage(getPage(), getPageSize(), fetchSize, parameterValues));
}

1
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java
View File

@@ -23,6 +23,7 @@ public class HibernateCursorItemReaderCommonTests extends AbstractDatabaseItemSt
reader.setQueryString(hsqlQuery);
reader.setSessionFactory(sessionFactory);
reader.setUseStatelessSession(true);
reader.setFetchSize(10);
reader.afterPropertiesSet();
reader.setSaveState(true);