diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java new file mode 100644 index 000000000..18ba8f8a8 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java @@ -0,0 +1,119 @@ +package org.springframework.batch.item.database; + +import org.springframework.batch.item.support.AbstractItemReaderItemStream; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.ClassUtils; +import org.springframework.util.Assert; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import java.util.List; + +/** + * Abstract {@link org.springframework.batch.item.ItemReader} for to extend when reading database records in a paging + * fashion. + * + * Implementations should execute queries using paged requests of a size specified in {@link #setPageSize(int)}. + * Additional pages are requested when needed as {@link #read()} method is called, returning an + * object corresponding to current position. + * + * @author Thomas Risberg + * @since 2.0 + */ +public abstract class AbstractPagingItemReader extends AbstractItemReaderItemStream implements InitializingBean { + + protected Log logger = LogFactory.getLog(getClass()); + + protected boolean initialized = false; + + protected int current = 0; + + protected int page = 0; + + protected int pageSize = 10; + + protected List results; + + public AbstractPagingItemReader() { + setName(ClassUtils.getShortName(AbstractPagingItemReader.class)); + } + + /** + * The number of rows to retreive at a time. + * + * @param pageSize the number of rows to fetch per page + */ + public void setPageSize(int pageSize) { + this.pageSize = pageSize; + } + + /** + * Check mandatory properties. + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.isTrue(pageSize > 0, "pageSize must be greater than zero"); + } + + @Override + protected T doRead() throws Exception { + + if (results == null || current >= pageSize) { + + if (logger.isDebugEnabled()) { + logger.debug("Reading page " + page); + } + + doReadPage(); + + if (current >= pageSize) { + current = 0; + } + page++; + } + + if (current < results.size()) { + return results.get(current++); + } + else { + return null; + } + + } + + abstract protected void doReadPage(); + + @Override + protected void doOpen() throws Exception { + + Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); + + initialized = true; + + } + + @Override + protected void doClose() throws Exception { + + initialized = false; + + } + + + @Override + protected void jumpToItem(int itemIndex) throws Exception { + + page = itemIndex / pageSize; + current = itemIndex % pageSize; + + doJumpToPage(itemIndex); + + if (logger.isDebugEnabled()) { + logger.debug("Jumping to page " + page + " and index " + current); + } + + } + + abstract protected void doJumpToPage(int itemIndex); + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java new file mode 100644 index 000000000..362140953 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java @@ -0,0 +1,75 @@ +package org.springframework.batch.item.database; + +import org.springframework.util.ClassUtils; +import org.springframework.util.Assert; +import org.springframework.orm.ibatis.SqlMapClientTemplate; + +import java.util.List; +import java.util.Map; +import java.util.HashMap; + +import com.ibatis.sqlmap.client.SqlMapClient; + +/** + * {@link org.springframework.batch.item.ItemReader} for reading database records using iBATIS in a paging + * fashion. + * + * It executes the query specified as the {@link #setQueryId(String)} to retrieve requested data. + * The query is executed using paged requests of a size specified in {@link #setPageSize(int)}. + * Additional pages are requested when needed as {@link #read()} method is called, returning an + * object corresponding to current position. + * + * The performance of the paging depends on the iBATIS implementation. + * + * Setting a fairly large page size and using a commit interval that matches the page size should provide + * better performance. + * + * The implementation is *not* thread-safe. + * + * @author Thomas Risberg + * @since 2.0 + */ +public class IbatisPagingItemReader extends AbstractPagingItemReader { + + private SqlMapClient sqlMapClient; + + private String queryId; + + private SqlMapClientTemplate sqlMapClientTemplate; + + public IbatisPagingItemReader() { + setName(ClassUtils.getShortName(IbatisPagingItemReader.class)); + } + + public void setSqlMapClient(SqlMapClient sqlMapClient) { + this.sqlMapClient = sqlMapClient; + } + + public void setQueryId(String queryId) { + this.queryId = queryId; + } + + /** + * Check mandatory properties. + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + super.afterPropertiesSet(); + Assert.notNull(sqlMapClient); + sqlMapClientTemplate = new SqlMapClientTemplate(sqlMapClient); + Assert.notNull(queryId); + } + + @Override + @SuppressWarnings("unchecked") + protected void doReadPage() { + //TODO: add support for parameter map + Map parameters = new HashMap(); + results = sqlMapClientTemplate.queryForList(queryId, parameters, (page * pageSize), pageSize); + } + + @Override + protected void doJumpToPage(int itemIndex) { + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java index 8be80961e..e0a4189a6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java @@ -15,7 +15,6 @@ */ package org.springframework.batch.item.database; -import org.springframework.batch.item.support.AbstractItemReaderItemStream; import org.springframework.batch.item.database.support.PagingQueryProvider; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.ClassUtils; @@ -29,7 +28,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import javax.sql.DataSource; -import java.util.List; import java.util.ArrayList; import java.sql.ResultSet; import java.sql.SQLException; @@ -54,7 +52,7 @@ import java.sql.SQLException; * @author Thomas Risberg * @since 2.0 */ -public class JdbcPagingItemReader extends AbstractItemReaderItemStream implements InitializingBean { +public class JdbcPagingItemReader extends AbstractPagingItemReader implements InitializingBean { protected Log logger = LogFactory.getLog(getClass()); @@ -70,18 +68,8 @@ public class JdbcPagingItemReader extends AbstractItemReaderItemStream imp private String remainingPagesSql; - private boolean initialized = false; - - private int current = 0; - - private int page = 0; - - private int pageSize = 10; - private Object startAfterValue; - private List results; - public JdbcPagingItemReader() { setName(ClassUtils.getShortName(JdbcPagingItemReader.class)); } @@ -94,15 +82,6 @@ public class JdbcPagingItemReader extends AbstractItemReaderItemStream imp this.queryProvider = queryProvider; } - /** - * The number of rows to retreive at a time. - * - * @param pageSize the number of rows to fetch per page - */ - public void setPageSize(int pageSize) { - this.pageSize = pageSize; - } - /** * The row mapper implementation to be used by this reader * @@ -117,8 +96,8 @@ public class JdbcPagingItemReader extends AbstractItemReaderItemStream imp * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { + super.afterPropertiesSet(); Assert.notNull(dataSource); - Assert.isTrue(pageSize > 0, "pageSize must be greater than zero"); JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); jdbcTemplate.setMaxRows(pageSize); this.simpleJdbcTemplate = new SimpleJdbcTemplate(jdbcTemplate); @@ -129,88 +108,50 @@ public class JdbcPagingItemReader extends AbstractItemReaderItemStream imp } @Override - @SuppressWarnings("unchecked") - protected T doRead() throws Exception { + protected void doReadPage() { + //TODO: add support for parameter map - if (results == null || current >= pageSize) { - - if (results == null) { - results = new ArrayList(); - } - else { - results.clear(); - } - - if (page == 0) { - if (logger.isDebugEnabled()) { - logger.debug("SQL used for reading first page: [" + firstPageSql + "]"); - } - simpleJdbcTemplate.getJdbcOperations().query(firstPageSql, - new RowCallbackHandler() { - public void processRow(ResultSet rs) throws SQLException { - startAfterValue = rs.getObject(1); - results.add(parameterizedRowMapper.mapRow(rs, results.size())); - } - }); - } - else { - if (logger.isDebugEnabled()) { - logger.debug("SQL used for reading remaining pages: [" + remainingPagesSql + "]"); - } - simpleJdbcTemplate.getJdbcOperations().query(remainingPagesSql, - new Object[] {startAfterValue}, - new RowCallbackHandler() { - public void processRow(ResultSet rs) throws SQLException { - startAfterValue = rs.getObject(1); - results.add(parameterizedRowMapper.mapRow(rs, results.size())); - } - }); - } - - if (current >= pageSize) { - current = 0; - } - page++; - } - - if (current < results.size()) { - return results.get(current++); + if (results == null) { + results = new ArrayList(); } else { - return null; + results.clear(); + } + + if (page == 0) { + if (logger.isDebugEnabled()) { + logger.debug("SQL used for reading first page: [" + firstPageSql + "]"); + } + simpleJdbcTemplate.getJdbcOperations().query(firstPageSql, + new RowCallbackHandler() { + public void processRow(ResultSet rs) throws SQLException { + startAfterValue = rs.getObject(1); + results.add(parameterizedRowMapper.mapRow(rs, results.size())); + } + }); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("SQL used for reading remaining pages: [" + remainingPagesSql + "]"); + } + simpleJdbcTemplate.getJdbcOperations().query(remainingPagesSql, + new Object[] {startAfterValue}, + new RowCallbackHandler() { + public void processRow(ResultSet rs) throws SQLException { + startAfterValue = rs.getObject(1); + results.add(parameterizedRowMapper.mapRow(rs, results.size())); + } + }); } } @Override - protected void doOpen() throws Exception { - - Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); - - initialized = true; - - } - - @Override - protected void doClose() throws Exception { - - initialized = false; - - } - - - @Override - protected void jumpToItem(int itemIndex) throws Exception { - - page = itemIndex / pageSize; - current = itemIndex % pageSize; - - logger.debug("Jumping to page " + page + " and index " + current); + protected void doJumpToPage(int itemIndex) { if (page > 0) { String jumpToItemSql; - jumpToItemSql = queryProvider.generateJumpToItemQuery(itemIndex, pageSize); if (logger.isDebugEnabled()) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java index 71207355e..375469d64 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java @@ -17,7 +17,6 @@ package org.springframework.batch.item.database; import java.util.HashMap; -import java.util.List; import java.util.Map; import javax.persistence.EntityManager; @@ -27,8 +26,6 @@ import javax.persistence.Query; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.batch.item.support.AbstractItemReaderItemStream; -import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -59,7 +56,7 @@ import org.springframework.util.ClassUtils; * @author Thomas Risberg * @since 2.0 */ -public class JpaPagingItemReader extends AbstractItemReaderItemStream implements InitializingBean { +public class JpaPagingItemReader extends AbstractPagingItemReader { protected Log logger = LogFactory.getLog(getClass()); @@ -69,16 +66,6 @@ public class JpaPagingItemReader extends AbstractItemReaderItemStream impl private String queryString; - private boolean initialized = false; - - private int current = 0; - - private int page = 0; - - private int pageSize = 10; - - private List entities; - public JpaPagingItemReader() { setName(ClassUtils.getShortName(JpaPagingItemReader.class)); } @@ -88,9 +75,9 @@ public class JpaPagingItemReader extends AbstractItemReaderItemStream impl } public void afterPropertiesSet() throws Exception { + super.afterPropertiesSet(); Assert.notNull(entityManagerFactory); Assert.hasLength(queryString); - Assert.isTrue(pageSize > 0, "pageSize must be greater than zero"); } /** @@ -100,80 +87,34 @@ public class JpaPagingItemReader extends AbstractItemReaderItemStream impl this.queryString = queryString; } - /** - * The number of entities to retrieve at a time. - * - * @param pageSize the number of entities per page - */ - public void setPageSize(int pageSize) { - this.pageSize = pageSize; - } - @Override @SuppressWarnings("unchecked") - protected T doRead() throws Exception { + protected void doReadPage() { + //TODO: add support for parameter map - if (entities == null || current >= pageSize) { - EntityManager entityManager = - entityManagerFactory.createEntityManager(jpaPropertyMap); - if (entityManager == null) { - throw new DataAccessResourceFailureException("Unable to obtain an EntityManager"); - } - - EntityTransaction tx = entityManager.getTransaction(); - tx.begin(); - - Query query = entityManager.createQuery(queryString) - .setFirstResult(page * pageSize) - .setMaxResults(pageSize); - - entities = query.getResultList(); - - entityManager.flush(); - entityManager.clear(); - - tx.commit(); - - if (current >= pageSize) { - current = 0; - } - page++; + EntityManager entityManager = + entityManagerFactory.createEntityManager(jpaPropertyMap); + if (entityManager == null) { + throw new DataAccessResourceFailureException("Unable to obtain an EntityManager"); } - if (current < entities.size()) { - return entities.get(current++); - } - else { - return null; - } + EntityTransaction tx = entityManager.getTransaction(); + tx.begin(); + Query query = entityManager.createQuery(queryString) + .setFirstResult(page * pageSize) + .setMaxResults(pageSize); + + results = query.getResultList(); + + entityManager.flush(); + entityManager.clear(); + + tx.commit(); } @Override - protected void doOpen() throws Exception { - - Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); - - initialized = true; - - } - - @Override - protected void doClose() throws Exception { - - initialized = false; - - } - - - @Override - protected void jumpToItem(int itemIndex) throws Exception { - - page = itemIndex / pageSize; - current = itemIndex % pageSize; - - logger.debug("Jumping to page " + page + " and index " + current); - + protected void doJumpToPage(int itemIndex) { } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/IbatisPagingItemReaderCommonTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/IbatisPagingItemReaderCommonTests.java new file mode 100644 index 000000000..da385c686 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/IbatisPagingItemReaderCommonTests.java @@ -0,0 +1,52 @@ +package org.springframework.batch.item.database; + +import org.junit.runner.RunWith; +import org.junit.internal.runners.JUnit4ClassRunner; +import org.springframework.batch.item.sample.Foo; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.database.support.IbatisKeyCollector; +import org.springframework.orm.ibatis.SqlMapClientFactoryBean; +import org.springframework.core.io.ClassPathResource; +import com.ibatis.sqlmap.client.SqlMapClient; + +@SuppressWarnings("unchecked") +@RunWith(JUnit4ClassRunner.class) +public class IbatisPagingItemReaderCommonTests extends CommonDatabaseItemStreamItemReaderTests { + + protected ItemReader getItemReader() throws Exception { + SqlMapClientFactoryBean factory = new SqlMapClientFactoryBean(); + factory.setConfigLocation(new ClassPathResource("ibatis-config.xml", getClass())); + factory.setDataSource(getDataSource()); + factory.afterPropertiesSet(); + SqlMapClient sqlMapClient = createSqlMapClient(); + + IbatisPagingItemReader reader = new IbatisPagingItemReader(); + reader.setQueryId("getAllFoos"); + reader.setSqlMapClient(sqlMapClient); + reader.setSaveState(true); + + reader.afterPropertiesSet(); + + return reader; + } + + private SqlMapClient createSqlMapClient() throws Exception { + SqlMapClientFactoryBean factory = new SqlMapClientFactoryBean(); + factory.setConfigLocation(new ClassPathResource("ibatis-config.xml", getClass())); + factory.setDataSource(getDataSource()); + factory.afterPropertiesSet(); + return (SqlMapClient) factory.getObject(); + } + + protected void pointToEmptyInput(ItemReader tested) throws Exception { + IbatisPagingItemReader reader = (IbatisPagingItemReader) tested; + reader.close(new ExecutionContext()); + + reader.setQueryId("getNoFoos"); + reader.afterPropertiesSet(); + + reader.open(new ExecutionContext()); + } + +}