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

BATCH-760: first cut of IbatisPagingItemReader, added AbstractPagingItemReader with common logic for all PagingItemReaders

Browse Source
This commit is contained in:
trisberg
2008-08-29 23:14:02 +00:00
parent c5a458f178
commit ab6068e5b6
5 changed files with 301 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java Normal file
View File

@@ -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<T> extends AbstractItemReaderItemStream<T> 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<T> 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);
}

75
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java Normal file
View File

@@ -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<T> extends AbstractPagingItemReader<T> {
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<String, Object> parameters = new HashMap<String, Object>();
results = sqlMapClientTemplate.queryForList(queryId, parameters, (page * pageSize), pageSize);
}
@Override
protected void doJumpToPage(int itemIndex) {
}
}

127
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
View File

@@ -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<T> extends AbstractItemReaderItemStream<T> implements InitializingBean {
public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> implements InitializingBean {
protected Log logger = LogFactory.getLog(getClass());
@@ -70,18 +68,8 @@ public class JdbcPagingItemReader<T> extends AbstractItemReaderItemStream<T> 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<T> results;
public JdbcPagingItemReader() {
setName(ClassUtils.getShortName(JdbcPagingItemReader.class));
}
@@ -94,15 +82,6 @@ public class JdbcPagingItemReader<T> extends AbstractItemReaderItemStream<T> 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<T> extends AbstractItemReaderItemStream<T> 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<T> extends AbstractItemReaderItemStream<T> 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<T>();
}
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()) {

101
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java
View File

@@ -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<T> extends AbstractItemReaderItemStream<T> implements InitializingBean {
public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
protected Log logger = LogFactory.getLog(getClass());
@@ -69,16 +66,6 @@ public class JpaPagingItemReader<T> extends AbstractItemReaderItemStream<T> impl
private String queryString;
private boolean initialized = false;
private int current = 0;
private int page = 0;
private int pageSize = 10;
private List<T> entities;
public JpaPagingItemReader() {
setName(ClassUtils.getShortName(JpaPagingItemReader.class));
}
@@ -88,9 +75,9 @@ public class JpaPagingItemReader<T> extends AbstractItemReaderItemStream<T> 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<T> extends AbstractItemReaderItemStream<T> 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) {
}
}