From ee0dcdd37a84138be4b95c1a2882c9e46c00865a Mon Sep 17 00:00:00 2001 From: lucasward Date: Mon, 15 Oct 2007 05:23:23 +0000 Subject: [PATCH] BATCH 144 and BATCH 150: create Hibernate and iBatis input sources with sample jobs. --- infrastructure/pom.xml | 50 +- .../orm/hibernate/HibernateInputSource.java | 143 ++ .../ibatis/IbatisDrivingQueryInputSource.java | 98 ++ .../io/orm/ibatis/IbatisInputSource.java | 39 + .../sql/CompositeKeyRestartDataConverter.java | 45 + ...ompositeKeySqlDrivingQueryInputSource.java | 135 ++ .../SingleKeySqlDrivingQueryInputSource.java | 219 ---- .../batch/io/sql/SqlCursorInputSource.java | 1150 ++++++++--------- .../AbstractDrivingQueryInputSource.java | 194 +++ .../HibernateInputSourceIntegrationTests.java | 36 + .../IbatisInputSourceIntegrationTests.java | 36 + .../batch/io/sample/domain/Foo.java | 54 + ...bstractSqlInputSourceIntegrationTests.java | 183 +++ .../batch/io/sql/CompositeKeyFooDao.java | 58 + ...ivingQueryInputSourceIntegrationTests.java | 81 ++ .../springframework/batch/io/sql/FooDao.java | 30 + .../batch/io/sql/FooInputSource.java | 48 + .../batch/io/sql/FooRowMapper.java | 21 + .../batch/io/sql/SingleKeyFooDao.java | 28 + ...ivingQueryInputSourceIntegrationTests.java | 152 --- .../SqlCursorInputSourceIntegrationTests.java | 333 +---- ...DataSourceInputSourceIntegrationTests.java | 183 +++ .../batch/io/orm/hibernate/Foo-write.hbm.xml | 15 + .../batch/io/orm/hibernate/Foo.hbm.xml | 15 + .../batch/io/orm/ibatis/ibatis-config.xml | 12 + .../batch/io/orm/ibatis/ibatis-foo.xml | 41 + .../io/sql/destroy-foo-schema-hsqldb.sql | 3 +- .../batch/io/sql/init-foo-schema-hsqldb.sql | 8 + 28 files changed, 2127 insertions(+), 1283 deletions(-) create mode 100644 infrastructure/src/main/java/org/springframework/batch/io/orm/hibernate/HibernateInputSource.java create mode 100644 infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisDrivingQueryInputSource.java create mode 100644 infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisInputSource.java create mode 100644 infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeyRestartDataConverter.java create mode 100644 infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSource.java delete mode 100644 infrastructure/src/main/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSource.java create mode 100644 infrastructure/src/main/java/org/springframework/batch/io/support/AbstractDrivingQueryInputSource.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/orm/hibernate/HibernateInputSourceIntegrationTests.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/orm/ibatis/IbatisInputSourceIntegrationTests.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sample/domain/Foo.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeyFooDao.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSourceIntegrationTests.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/FooDao.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/FooInputSource.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/FooRowMapper.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeyFooDao.java delete mode 100644 infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSourceIntegrationTests.java create mode 100644 infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceInputSourceIntegrationTests.java create mode 100644 infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo-write.hbm.xml create mode 100644 infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo.hbm.xml create mode 100644 infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-config.xml create mode 100644 infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-foo.xml diff --git a/infrastructure/pom.xml b/infrastructure/pom.xml index 42eda75ea..5ee64fe6b 100644 --- a/infrastructure/pom.xml +++ b/infrastructure/pom.xml @@ -1,6 +1,4 @@ - + 4.0.0 spring-batch-infrastructure @@ -102,20 +100,44 @@ true - org.springframework.ws + org.springframework.ws spring-oxm - 1.0.0 - true - - + 1.0.0 + true + + org.springframework spring-beans - - + + org.springframework spring-core - - + + + + + org.hibernate + hibernate + true + + + + org.apache.geronimo.specs + geronimo-jta_1.1_spec + 1.1 + provided + + + + cglib + cglib-nodep + true + + + org.apache.ibatis + ibatis-sqlmap + 2.3.0 + true @@ -125,9 +147,7 @@ org.apache.maven.plugins maven-clover-plugin - - ${basedir}/src/test/resources/clover.license - + ${basedir}/src/test/resources/clover.license diff --git a/infrastructure/src/main/java/org/springframework/batch/io/orm/hibernate/HibernateInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/orm/hibernate/HibernateInputSource.java new file mode 100644 index 000000000..879cd6cca --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/orm/hibernate/HibernateInputSource.java @@ -0,0 +1,143 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.orm.hibernate; + +import java.util.Properties; + +import org.hibernate.ScrollableResults; +import org.hibernate.SessionFactory; +import org.hibernate.StatelessSession; +import org.springframework.batch.io.InputSource; +import org.springframework.batch.item.ResourceLifecycle; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.restart.GenericRestartData; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationAdapter; +import org.springframework.util.Assert; + +/** + * {@link InputSource} for reading database records built on top of Hibernate. + * + * @author Robert Kasanicky + */ +public class HibernateInputSource implements InputSource, Restartable, InitializingBean, DisposableBean, + ResourceLifecycle { + + private static final String RESTART_DATA_ROW_NUMBER_KEY = "HibernateInputSource.rowNumber"; + + private SessionFactory sessionFactory; + private StatelessSession session; + private ScrollableResults cursor; + private String queryString; + + private int lastCommitRowNumber = 0; + + private boolean initialized = false; + private TransactionSynchronization synchronization = new HibernateInputSourceTransactionSynchronization(); + + + public Object read() { + if (!initialized) { + open(); + } + if (cursor.next()) { + return cursor.get(0); + } + return null; + } + + /** + * Close the resultset cursor and hibernate session. + */ + public void close() { + initialized = false; + cursor.close(); + session.close(); + } + + /** + * Create cursor for the query + */ + public void open() { + session = sessionFactory.openStatelessSession(); + cursor = session.createQuery(queryString).scroll(); + + BatchTransactionSynchronizationManager.registerSynchronization(synchronization ); + initialized = true; + } + + public void setSessionFactory(SessionFactory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + public void afterPropertiesSet() throws Exception { + Assert.notNull(sessionFactory); + Assert.hasLength(queryString); + } + + public void destroy() throws Exception { + close(); + } + + public void setQueryString(String queryString) { + this.queryString = queryString; + } + + /** + * @return the current row number wrapped as RestartData + */ + public RestartData getRestartData() { + Properties props = new Properties(); + props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, String.valueOf(cursor.getRowNumber())); + + return new GenericRestartData(props); + } + + /** + * Set the cursor to the received row number. + */ + public void restoreFrom(RestartData data) { + Assert.state(!initialized, + "Cannot restore when already intialized. Call close() first before restore()"); + + Properties props = data.getProperties(); + if (props.getProperty(RESTART_DATA_ROW_NUMBER_KEY) == null) { + return; + } + int rowNumber = Integer.parseInt(props.getProperty(RESTART_DATA_ROW_NUMBER_KEY)); + open(); + cursor.setRowNumber(rowNumber); + } + + /** + * Encapsulates transaction events handling. + */ + private class HibernateInputSourceTransactionSynchronization extends TransactionSynchronizationAdapter { + + public void afterCompletion(int status) { + if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + cursor.setRowNumber(lastCommitRowNumber); + } else if (status == TransactionSynchronization.STATUS_COMMITTED) { + lastCommitRowNumber = cursor.getRowNumber(); + } + } + } + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisDrivingQueryInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisDrivingQueryInputSource.java new file mode 100644 index 000000000..ab97f307f --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisDrivingQueryInputSource.java @@ -0,0 +1,98 @@ +package org.springframework.batch.io.orm.ibatis; + +import java.util.List; +import java.util.Properties; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.sql.SingleKeySqlDrivingQueryInputSource; +import org.springframework.batch.io.support.AbstractDrivingQueryInputSource; +import org.springframework.batch.restart.GenericRestartData; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.orm.ibatis.SqlMapClientTemplate; +import org.springframework.util.Assert; + +import com.ibatis.sqlmap.client.SqlMapClient; + +/** + * Driving query {@link InputSource} based on iBATIS ORM framework. It is functionally similar to + * {@link SingleKeySqlDrivingQueryInputSource} but does not make assumptions about the primary key + * structure. + * + * @see SingleKeySqlDrivingQueryInputSource + * + * @author Robert Kasanicky + * @author Lucas Ward + */ +public class IbatisDrivingQueryInputSource extends AbstractDrivingQueryInputSource implements Restartable { + + public static final String RESTART_KEY = "IbatisDrivingQueryInputSource.keyIndex"; + + private SqlMapClientTemplate sqlMapClientTemplate; + + private String drivingQuery; + + private String restartQueryId; + + protected List retrieveKeys() { + return sqlMapClientTemplate.queryForList(drivingQuery); + } + + + public RestartData getRestartData() { + Properties props = new Properties(); + props.setProperty(RESTART_KEY, getCurrentKey().toString()); + + return new GenericRestartData(props); + } + + /** + * Restore the keys list given the provided restart data. + * + * @see org.springframework.batch.io.support.AbstractDrivingQueryInputSource#restoreKeys(org.springframework.batch.restart.RestartData) + */ + public List restoreKeys(RestartData data) { + + Properties props = data.getProperties(); + Object key = props.getProperty(RESTART_KEY); + return sqlMapClientTemplate.queryForList(restartQueryId, key); + } + + /* (non-Javadoc) + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(sqlMapClientTemplate, "SqlMaperClientTemplate must not be null."); + Assert.hasText(drivingQuery, "The DrivingQuery must not be null or empty."); + } + + /** + * @param sqlMapClient configured iBATIS client + */ + public void setSqlMapClient(SqlMapClient sqlMapClient) { + this.sqlMapClientTemplate = new SqlMapClientTemplate(); + this.sqlMapClientTemplate.setSqlMapClient(sqlMapClient); + } + + /** + * @param drivingQueryId id of the iBATIS select statement that will be used + * to retrieve the list of primary keys + */ + public void setDrivingQueryId(String drivingQueryId) { + this.drivingQuery = drivingQueryId; + } + + /** + * Set the id of the restart query. + * + * @param restartQueryId id of the iBatis select statement that will be used + * to retrieve the list of primary keys after a restart. + */ + public void setRestartQueryId(String restartQueryId) { + this.restartQueryId = restartQueryId; + } + + public final SqlMapClientTemplate getSqlMapClientTemplate() { + return sqlMapClientTemplate; + } +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisInputSource.java new file mode 100644 index 000000000..a7fa7979a --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisInputSource.java @@ -0,0 +1,39 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.orm.ibatis; + +/** + * @author Lucas Ward + * + */ +public class IbatisInputSource extends IbatisDrivingQueryInputSource { + + String detailsQueryId; + + public Object read() { + return getSqlMapClientTemplate().queryForObject(detailsQueryId, super.read()); + } + + + /** + * @param detailsQueryId id of the iBATIS select statement that will used + * to retrieve an object for a single primary key from the list + * returned by driving query + */ + public void setDetailsQueryId(String detailsQueryId) { + this.detailsQueryId = detailsQueryId; + } +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeyRestartDataConverter.java b/infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeyRestartDataConverter.java new file mode 100644 index 000000000..b8099363b --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeyRestartDataConverter.java @@ -0,0 +1,45 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.sql; + +import org.springframework.batch.restart.RestartData; + +/** + * Converts an object representing a composite key to RestartData and + * back again. + * + * @author Lucas Ward + * + */ +public interface CompositeKeyRestartDataConverter { + + /** + * Given the provided composite key, return a RestartData representation. + * + * @param compositeKey + * @return ResartData representing the composite key. + */ + public RestartData createRestartData(Object compositeKey); + + /** + * Given the provided restart data, return an array of objects that can + * be used as parameters to a driving query. + * + * @param restartData + * @return an array of objects that can be used as arguments to a JdbcTemplate. + */ + public Object[] createArguments(RestartData restartData); +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSource.java new file mode 100644 index 000000000..ebedbb06a --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSource.java @@ -0,0 +1,135 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.sql; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.io.support.AbstractDrivingQueryInputSource; +import org.springframework.batch.restart.RestartData; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + *

Sql implementation of the DrivingQueryInputSource that works for composite keys. + * (i.e. keys represented by multiple columns) A sql query to be used to return the keys and + * a RowMapper to map each row in the resultset to an Object must be set in order for the + * InputSource to work correctly. + *

+ * + * @author Lucas Ward + * @see AbstractDrivingQueryInputSource + */ +public class CompositeKeySqlDrivingQueryInputSource extends + AbstractDrivingQueryInputSource { + + public static final String RESTART_KEY = "CompositeKeySqlDrivingQueryInputSource.key"; + + JdbcTemplate jdbcTemplate; + + RowMapper keyMapper; + + String drivingQuery; + + String restartQuery; + + CompositeKeyRestartDataConverter restartDataConverter; + + public CompositeKeySqlDrivingQueryInputSource() { + super(); + } + + /** + * Construct a new InputSource. + * + * @param jdbcTemplate + * @param drivingQuery - Sql statement that returns all keys to process. + * @param keyMapper - RowMapper that maps each row of the ResultSet to an object. + */ + public CompositeKeySqlDrivingQueryInputSource(JdbcTemplate jdbcTemplate, + String drivingQuery, RowMapper keyMapper){ + this(); + Assert.notNull(jdbcTemplate, "The JdbcTemplate must not be null."); + Assert.hasText(drivingQuery, "The DrivingQuery must not be null or empty."); + Assert.notNull(keyMapper, "The key RowMapper must not be null."); + this.jdbcTemplate = jdbcTemplate; + this.drivingQuery = drivingQuery; + this.keyMapper = keyMapper; + } + + /* (non-Javadoc) + * @see org.springframework.batch.io.sql.scratch.AbstractDrivingQueryInputSource#retrieveKeys() + */ + protected List retrieveKeys() { + return jdbcTemplate.query(drivingQuery, keyMapper); + } + + /* (non-Javadoc) + * @see org.springframework.batch.io.sql.scratch.AbstractDrivingQueryInputSource#restoreKeys(org.springframework.batch.restart.RestartData) + */ + protected List restoreKeys(RestartData restartData) { + + Assert.state(restartDataConverter != null, "RestartDataConverter must not be null."); + Assert.state(StringUtils.hasText(restartQuery), "The RestartQuery must not be null or empty" + + " in order to restart."); + + if (restartData.getProperties() != null) { + return jdbcTemplate.query(restartQuery, restartDataConverter.createArguments(restartData), keyMapper); + } + + return new ArrayList(); + } + + /* (non-Javadoc) + * @see org.springframework.batch.restart.Restartable#getRestartData() + */ + public RestartData getRestartData() { + Assert.state(restartDataConverter != null, "RestartDataConverter must not be null."); + return restartDataConverter.createRestartData(getCurrentKey()); + } + + /** + * Set the {@link RestartDataConverter} used to convert a composite key to + * RestartData and back again. + * + * @param restartDataConverter + */ + public void setRestartDataConverter( + CompositeKeyRestartDataConverter restartDataConverter) { + this.restartDataConverter = restartDataConverter; + } + + /** + * Set the query to use to retrieve keys in order to restore the previous + * state for restart. + * + * @param restartQuery + */ + public void setRestartQuery(String restartQuery) { + this.restartQuery = restartQuery; + } + + /* (non-Javadoc) + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(jdbcTemplate, "The JdbcTemplate must not be null."); + Assert.hasText(drivingQuery, "The DrivingQuery must not be null or empty."); + Assert.notNull(keyMapper, "The key RowMapper must not be null."); + } +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSource.java deleted file mode 100644 index d188b1435..000000000 --- a/infrastructure/src/main/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSource.java +++ /dev/null @@ -1,219 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.batch.io.sql; - -import java.sql.ResultSet; -import java.sql.SQLException; -import java.util.Iterator; -import java.util.List; -import java.util.Properties; - -import javax.sql.DataSource; - -import org.springframework.batch.io.InputSource; -import org.springframework.batch.item.ResourceLifecycle; -import org.springframework.batch.restart.GenericRestartData; -import org.springframework.batch.restart.RestartData; -import org.springframework.batch.restart.Restartable; -import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.core.RowMapper; -import org.springframework.util.Assert; - -/** - *

- * DrivingQuery based input source for input data that can be uniquely - * identified by a single primary key. The current implementation is - * forward-only, and requires a transactional buffer to ensure rollbacks are - * handled correctly. - *

- * - *

- * Users of this input source must provide a 'Driving Query' that returns only - * one column. (If more than one is returned, the first column will be used) A - * 'Details Query' must then be provided that requires only one parameter. - * (question mark) Invalid queries will throw SqlExceptions from JdbcTemplate. - *

- * - * - * @author Lucas Ward - * - */ -public class SingleKeySqlDrivingQueryInputSource implements ResourceLifecycle, InputSource, Restartable { - - private static final String RESTART_KEY = "SingleKeySqlDrivingQueryInputSource.lastProcessedKey"; - - private DataSource dataSource; - - private JdbcTemplate jdbcTemplate; - - private String drivingQuery; - - private String detailsQuery; - - private String restartQuery; - - private Object[] detailArgs = new Object[1]; - - private List keys; - - private Iterator keysIterator; - - private RowMapper mapper; - - /** - * Read one record by passing in the current key to the details query. - * - */ - public Object read() { - if (keys == null) { - retrieveKeys(); - } - - if (keysIterator.hasNext()) { - detailArgs[0] = keysIterator.next(); - - return jdbcTemplate.queryForObject(detailsQuery, detailArgs, mapper); - } - - return null; - } - - /* - * Retrieve the keys by calling the DrivingQuery. - */ - private void retrieveKeys() { - - jdbcTemplate = new JdbcTemplate(dataSource); - - keys = jdbcTemplate.query(drivingQuery, new SingleColumnRowMapper()); - - keysIterator = keys.iterator(); - } - - /* - * (non-Javadoc) - * @see org.springframework.batch.item.ResourceLifecycle#close() - */ - public void close() { - keys = null; - keysIterator = null; - } - - /* - * (non-Javadoc) - * @see org.springframework.batch.item.ResourceLifecycle#open() - */ - public void open() { - - } - - public void setDataSource(DataSource dataSource) { - this.dataSource = dataSource; - } - - /** - * Set the query to be used to obtain the list of keys at - * initialization. Each key returned will be fed into the - * details query. - * - * @param drivingQuery - */ - public void setDrivingQuery(String drivingQuery) { - this.drivingQuery = drivingQuery; - } - - /** - * Set the query to be used for each 'detail' record. Meaning, - * the query each key (row returned from the driving query) will - * be fed into in order to return a row to be mapped. - * - * @param detailsQuery - */ - public void setDetailsQuery(String detailsQuery) { - this.detailsQuery = detailsQuery; - } - - /** - * Set the query to be used in the case of a restart. The current - * key at the time restart data is requested will be fed into this - * query as a parameter upon restart, allowing for only the remaining - * keys to be returned. - * - * @param restartQuery - */ - public void setRestartQuery(String restartQuery) { - this.restartQuery = restartQuery; - } - - /** - * Set RowMapper to be used for each call to the provided details - * query. - * - * @param mapper - */ - public void setMapper(RowMapper mapper) { - this.mapper = mapper; - } - - // Required because JdbcTemplate.queryForList returns a list - // of maps based on metadata. - private class SingleColumnRowMapper implements RowMapper { - - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - return rs.getObject(1); - } - - } - - public RestartData getRestartData() { - Properties props = new Properties(); - props.setProperty(RESTART_KEY, detailArgs[0].toString()); - return new GenericRestartData(props); - } - - /** - * Restore input source to previous state. If the input source has already - * been initialized before calling restore (meaning, read has been called) - * then an IllegalStateException will be thrown, since all input sources - * should be restored before being read from, otherwise already processed - * data could be returned. The RestartData attempting to be restored from - * must have been obtained from the same input source as the one - * being restored from otherwise it is invalid. - * - * @param RestartData obtained by calling getRestartData during a previous - * run. - * @throws IllegalStateException if input source has already been read from. - */ - public void restoreFrom(RestartData data) { - - Assert.notNull(data, "RestartData must not be null."); - - if (keys != null) { - throw new IllegalStateException("Cannot restore when already intialized. Call" - + " close() first before restore()"); - } - - Properties restartData = data.getProperties(); - String lastProcessedKey = restartData.getProperty(RESTART_KEY); - if (lastProcessedKey != null) { - jdbcTemplate = new JdbcTemplate(dataSource); - - keys = jdbcTemplate.query(restartQuery, new Object[] { lastProcessedKey }, new SingleColumnRowMapper()); - - keysIterator = keys.iterator(); - } - } -} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/sql/SqlCursorInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/sql/SqlCursorInputSource.java index 949423571..3bb6b3dc8 100644 --- a/infrastructure/src/main/java/org/springframework/batch/io/sql/SqlCursorInputSource.java +++ b/infrastructure/src/main/java/org/springframework/batch/io/sql/SqlCursorInputSource.java @@ -1,575 +1,575 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.io.sql; - -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.SQLWarning; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.List; -import java.util.Properties; - -import javax.sql.DataSource; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.batch.io.InputSource; -import org.springframework.batch.io.Skippable; -import org.springframework.batch.item.ResourceLifecycle; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.restart.GenericRestartData; -import org.springframework.batch.restart.RestartData; -import org.springframework.batch.restart.Restartable; -import org.springframework.batch.statistics.StatisticsProvider; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.dao.DataAccessException; -import org.springframework.dao.InvalidDataAccessResourceUsageException; -import org.springframework.jdbc.SQLWarningException; -import org.springframework.jdbc.core.RowMapper; -import org.springframework.jdbc.support.JdbcUtils; -import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator; -import org.springframework.jdbc.support.SQLExceptionTranslator; -import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationAdapter; -import org.springframework.util.Assert; - -/** - *

- * Simple input source that opens a Sql Cursor and continually retrieves the - * next row in the ResultSet. It is extremely important to note that the - * JdbcDriver used must be version 3.0 or higher. This is because earlier - * versions do not support holding a ResultSet open over commits. - *

- * - *

- * Each call to read() will call the provided RowMapper, (NOTE: Calling read() - * without setting a RowMapper will result in an IllegalStateException!) passing - * in the ResultSet. If this is the first call to read(), the provided query - * will be run in order to open the cursor. There is currently no wrapping of - * the ResultSet to suppress calls to next(). However, if the RowMapper - * increments the current row, the next call to read will verify that the - * current row is at the expected position and throw a DataAccessException if it - * is not. This means that, in theory, a RowMapper could read ahead, as long as - * it returns the row back to it's correct position before returning. The reason - * for such strictness on the ResultSet is due to the need to maintain strict - * control for Transactions, restartability and skippability. This ensures that - * each call to read() returns the ResultSet at the correct line, regardless of - * rollbacks, restarts, or skips. - *

- * - *

- * Restart: This implementation contains basic, simple restart. The current row - * is returned as restart data, and when restored from that same data, the - * cursor is opened and the current row set to the value within the restart - * data. - *

- * - *

- * Statistics: There are two statistics returned by this input source: the - * current line being processed and the number of lines that have been skipped. - *

- * - *

- * Transactions: At first glance, it may appear odd that Spring's - * TransactionSynchronization abstraction is used for something that is reading - * from the database, however, it is important because the same resultset is - * held open regardless of commits or roll backs. This means that when a - * transaction is committed, the input source is notified so that it can save - * it's current row number. Later, if the transaction is rolled back, the - * current row can be moved back to the same row number as it was on when commit - * was called. - *

- * - *

- * Calling skip will indicate to the input source that a record is bad and - * should not be represented to the user if the transaction is rolled back. For - * example, if row 2 is read in, and found to be bad, calling skip will inform - * the Input Source. If reading is then continued, and a rollback is necessary - * because of an error on output, the input source will be returned to row 1. - * Calling read while on row 1 will move the current row to 3, not 2, because 2 - * has been marked as skipped. - *

- * - *

- * Calling close on this Input Source will cause all resources it is currently - * using to be freed. (Connection, resultset, etc). If read() is called on the - * same instance again, the cursor will simply be reopened starting at row 0. - *

- * - * @author Lucas Ward - * @author Peter Zozom - */ -public class SqlCursorInputSource implements InputSource, ResourceLifecycle, DisposableBean, - InitializingBean, Restartable, StatisticsProvider, Skippable { - - private static Log log = LogFactory.getLog(SqlCursorInputSource.class); - - public static final int VALUE_NOT_SET = -1; - - private static final String CURRENT_PROCESSED_ROW = "sqlCursorInput.lastProcessedRowNum"; - - private static final String SKIP_COUNT = "sqlCursorInput.skippedRrecordCount"; - - private Connection con; - - private Statement stmt; - - protected ResultSet rs; - - private DataSource dataSource; - - private String sql; - - private final List skippedRows = new ArrayList(); - - private int skipCount = 0; - - private int fetchSize = VALUE_NOT_SET; - - private int maxRows = VALUE_NOT_SET; - - private int queryTimeout = VALUE_NOT_SET; - - private boolean ignoreWarnings = true; - - private boolean verifyCursorPosition = true; - - private SQLExceptionTranslator exceptionTranslator; - - /* Current count of processed records. */ - private int currentProcessedRow = 0; - - private int lastCommittedRow = 0; - - private final SqlInputTransactionSynchronization transactionSynchronization = new SqlInputTransactionSynchronization(); - - private RowMapper mapper; - - /** - * Assert that mandatory properties are set. - * - * @throws IllegalArgumentException - * if either data source or sql properties not set. - */ - public void afterPropertiesSet() throws Exception { - Assert.notNull(dataSource, "DataSOurce must be provided"); - Assert.notNull(sql, "The SQL query must be provided"); - } - - /** - * Public setter for the data source for injection purposes. - * - * @param dataSource - */ - public void setDataSource(DataSource dataSource) { - this.dataSource = dataSource; - } - - /** - * Increment the cursor to the next row, validating the cursor position and - * passing the resultset to the RowMapper. If read has not been called on - * this instance before, the cursor will be opened. If there are skipped - * records for this commit scope, an internal list of skipped records will - * be checked to ensure that only a valid row is given to the mapper. - * - * @returns Object returned by RowMapper - * @throws DataAccessException - * @throws IllegalStateExceptino - * if mapper is null. - */ - public Object read() { - - if (this.rs == null) { - this.executeQuery(); - } - - Assert.state(mapper != null, "Mapper must not be null."); - - try { - if (!rs.next()) { - return null; - } else { - currentProcessedRow++; - if (!skippedRows.isEmpty()) { - // while is necessary to handle successive skips. - while (skippedRows - .contains(new Integer(currentProcessedRow))) { - if (!rs.next()) { - return null; - } - currentProcessedRow++; - } - } - - Object mappedResult = mapper.mapRow(rs, currentProcessedRow); - - verifyCursorPosition(currentProcessedRow); - - return mappedResult; - } - } catch (SQLException se) { - throw getExceptionTranslator().translate( - "Trying to process next row", sql, se); - } - - } - - public int getCurrentProcessedRow() { - return currentProcessedRow; - } - - /** - * Mark the current row. Calling reset will cause the result set to be set - * to the current row when mark was called. - */ - private void mark() { - lastCommittedRow = currentProcessedRow; - skippedRows.clear(); - } - - /** - * Set the ResultSet's current row to the last marked position. - * - * @throws DataAccessException - */ - private void reset() { - try { - currentProcessedRow = lastCommittedRow; - if (currentProcessedRow > 0) { - rs.absolute(currentProcessedRow); - } else { - rs.beforeFirst(); - } - - } catch (SQLException se) { - throw getExceptionTranslator().translate( - "Attempted to move ResultSet to last committed row", sql, - se); - } - } - - /** - * Close this input source. The ResultSet, Statement and Connection created - * will be closed. This must be called or the connection and cursor will be - * held open indefinitely! - * - * @see org.springframework.batch.item.ResourceLifecycle#close() - */ - public void close() { - JdbcUtils.closeResultSet(this.rs); - JdbcUtils.closeStatement(this.stmt); - JdbcUtils.closeConnection(this.con); - this.currentProcessedRow = 0; - skippedRows.clear(); - skipCount = 0; - } - - /** - * Calls close to ensure that bean factories can close and always release - * resources. - * - * @see org.springframework.beans.factory.DisposableBean#destroy() - */ - public void destroy() throws Exception { - close(); - } - - // Check the result set is in synch with the currentRow attribute. This is - // important - // to ensure that the user hasn't modified the current row. - private void verifyCursorPosition(int expectedCurrentRow) - throws SQLException { - if (verifyCursorPosition) { - if (expectedCurrentRow != this.rs.getRow()) { - throw new InvalidDataAccessResourceUsageException( - "Unexpected cursor position change."); - } - } - } - - /* - * Executes the provided SQL query. The statement is created with - * 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' set to true. This is extremely - * important, since a non read-only cursor may lock tables that shouldn't be - * locked, and not holding the cursor open over a commit would require it to - * be reopened after each commit, which would destroy performance. - */ - private void executeQuery() { - - Assert.state(dataSource != null, "DataSource must not be null."); - - try { - this.con = dataSource.getConnection(); - this.stmt = this.con.createStatement( - ResultSet.TYPE_SCROLL_INSENSITIVE, - ResultSet.CONCUR_READ_ONLY, - ResultSet.HOLD_CURSORS_OVER_COMMIT); - applyStatementSettings(this.stmt); - this.rs = this.stmt.executeQuery(sql); - handleWarnings(this.stmt.getWarnings()); - } catch (SQLException se) { - close(); - throw getExceptionTranslator().translate("Executing query", - getSql(), se); - } - - BatchTransactionSynchronizationManager - .registerSynchronization(transactionSynchronization); - } - - /* - * Prepare the given JDBC Statement (or PreparedStatement or - * CallableStatement), applying statement settings such as fetch size, max - * rows, and query timeout. @param stmt the JDBC Statement to prepare - * @throws SQLException - * - * @see #setFetchSize - * @see #setMaxRows - * @see #setQueryTimeout - */ - private void applyStatementSettings(Statement stmt) throws SQLException { - if (fetchSize != VALUE_NOT_SET) { - stmt.setFetchSize(fetchSize); - stmt.setFetchDirection(ResultSet.FETCH_FORWARD); - } - if (maxRows != VALUE_NOT_SET) { - stmt.setMaxRows(maxRows); - } - if (queryTimeout != VALUE_NOT_SET) { - stmt.setQueryTimeout(queryTimeout); - } - } - - /* - * Return the exception translator for this instance.

Creates a default - * SQLErrorCodeSQLExceptionTranslator for the specified DataSource if none - * is set. - */ - protected SQLExceptionTranslator getExceptionTranslator() { - if (exceptionTranslator == null) { - if (dataSource != null) { - exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator( - dataSource); - } else { - exceptionTranslator = new SQLStateSQLExceptionTranslator(); - } - } - return exceptionTranslator; - } - - /* - * Throw a SQLWarningException if we're not ignoring warnings, else log the - * warnings (at debug level). - * - * @param warning the warnings object from the current statement. May be - * null, in which case this method does nothing. - * - * @see org.springframework.jdbc.SQLWarningException - */ - private void handleWarnings(SQLWarning warnings) throws SQLWarningException { - if (ignoreWarnings) { - SQLWarning warningToLog = warnings; - while (warningToLog != null) { - log.debug("SQLWarning ignored: SQL state '" - + warningToLog.getSQLState() + "', error code '" - + warningToLog.getErrorCode() + "', message [" - + warningToLog.getMessage() + "]"); - warningToLog = warningToLog.getNextWarning(); - } - } else if(warnings != null){ - throw new SQLWarningException("Warning not ignored", warnings); - } - } - - /* - * (non-Javadoc) - * - * @see org.springframework.batch.restart.Restartable#getRestartData() - */ - public RestartData getRestartData() { - return new GenericRestartData(getStatistics()); - } - - /* - * (non-Javadoc) - * - * @see org.springframework.batch.restart.Restartable#restoreFrom(org.springframework.batch.restart.RestartData) - */ - public void restoreFrom(RestartData data) { - if (data == null) - return; - - if (rs == null) { - executeQuery(); - } - - Properties restartProperties = data.getProperties(); - if(restartProperties.containsKey(CURRENT_PROCESSED_ROW) == false){ - return; - } - - try { - this.currentProcessedRow = Integer.parseInt(restartProperties - .getProperty(CURRENT_PROCESSED_ROW)); - rs.absolute(currentProcessedRow); - } catch (SQLException se) { - throw getExceptionTranslator().translate( - "Attempted to move ResultSet to last committed row", sql, - se); - } - } - - /* - * (non-Javadoc) - * - * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() - */ - public Properties getStatistics() { - - Properties props = new Properties(); - props.setProperty(CURRENT_PROCESSED_ROW, new Integer( - currentProcessedRow).toString()); - props.setProperty(SKIP_COUNT, new Integer(skipCount).toString()); - return props; - } - - /** - * Skip the current row. If the transaction is rolled back, this row will - * not be represented to the RowMapper when read() is called. For example, - * if you read in row 2, find the data to be bad, and call skip(), then - * continue processing and find - */ - public void skip() { - skippedRows.add(new Integer(currentProcessedRow)); - skipCount++; - } - - /** - * Gives the JDBC driver a hint as to the number of rows that should be - * fetched from the database when more rows are needed for this - * ResultSet object. If the fetch size specified is zero, the - * JDBC driver ignores the value. - * - * @param fetchSize - * the number of rows to fetch - * @see ResultSet#setFetchSize(int) - */ - public void setFetchSize(int fetchSize) { - this.fetchSize = fetchSize; - } - - /** - * Sets the limit for the maximum number of rows that any - * ResultSet object can contain to the given number. - * - * @param maxRows - * the new max rows limit; zero means there is no limit - * @see Statement#setMaxRows(int) - */ - public void setMaxRows(int maxRows) { - this.maxRows = maxRows; - } - - /** - * Sets the number of seconds the driver will wait for a - * Statement object to execute to the given number of - * seconds. If the limit is exceeded, an SQLException is - * thrown. - * - * @param queryTimeout - * seconds the new query timeout limit in seconds; zero means - * there is no limit - * @see Statement#setQueryTimeout(int) - */ - public void setQueryTimeout(int queryTimeout) { - this.queryTimeout = queryTimeout; - } - - /** - * Set whether SQLWarnings should be ignored (only logged) or exception - * should be thrown. - * - * @param ignoreWarnings - * if TRUE, warnings are ignored - */ - public void setIgnoreWarnings(boolean ignoreWarnings) { - this.ignoreWarnings = ignoreWarnings; - } - - /** - * Allow verification of cursor position after current row is processed by - * RowMapper or RowCallbackHandler. Default value is TRUE. - * - * @param verifyCursorPosition - * if true, cursor position is verified - */ - public void setVerifyCursorPosition(boolean verifyCursorPosition) { - this.verifyCursorPosition = verifyCursorPosition; - } - - /** - * Set the RowMapper to be used for all calls to read(). - * - * @param mapper - */ - public void setMapper(RowMapper mapper) { - this.mapper = mapper; - } - - /** - * Set the sql statement to be used when creating the cursor. This statement - * should be a complete and valid Sql statement, as it will be run directly - * without any modification. - * - * @param sql - */ - public void setSql(String sql) { - this.sql = sql; - } - - public String getSql() { - return sql; - } - - public void open() { - // TODO Auto-generated method stub - - } - - private class SqlInputTransactionSynchronization extends - TransactionSynchronizationAdapter { - - /* - * @param status transaction status - * - * @see org.springframework.transaction.support.TransactionSynchronization#afterCompletion(int) - */ - public void afterCompletion(int status) { - - if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { - reset(); - } else if (status == TransactionSynchronization.STATUS_COMMITTED) { - mark(); - } - } - } - -} +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.io.sql; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.SQLWarning; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.Skippable; +import org.springframework.batch.item.ResourceLifecycle; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.restart.GenericRestartData; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.batch.statistics.StatisticsProvider; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.InvalidDataAccessResourceUsageException; +import org.springframework.jdbc.SQLWarningException; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.support.JdbcUtils; +import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator; +import org.springframework.jdbc.support.SQLExceptionTranslator; +import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationAdapter; +import org.springframework.util.Assert; + +/** + *

+ * Simple input source that opens a Sql Cursor and continually retrieves the + * next row in the ResultSet. It is extremely important to note that the + * JdbcDriver used must be version 3.0 or higher. This is because earlier + * versions do not support holding a ResultSet open over commits. + *

+ * + *

+ * Each call to read() will call the provided RowMapper, (NOTE: Calling read() + * without setting a RowMapper will result in an IllegalStateException!) passing + * in the ResultSet. If this is the first call to read(), the provided query + * will be run in order to open the cursor. There is currently no wrapping of + * the ResultSet to suppress calls to next(). However, if the RowMapper + * increments the current row, the next call to read will verify that the + * current row is at the expected position and throw a DataAccessException if it + * is not. This means that, in theory, a RowMapper could read ahead, as long as + * it returns the row back to it's correct position before returning. The reason + * for such strictness on the ResultSet is due to the need to maintain strict + * control for Transactions, restartability and skippability. This ensures that + * each call to read() returns the ResultSet at the correct line, regardless of + * rollbacks, restarts, or skips. + *

+ * + *

+ * Restart: This implementation contains basic, simple restart. The current row + * is returned as restart data, and when restored from that same data, the + * cursor is opened and the current row set to the value within the restart + * data. + *

+ * + *

+ * Statistics: There are two statistics returned by this input source: the + * current line being processed and the number of lines that have been skipped. + *

+ * + *

+ * Transactions: At first glance, it may appear odd that Spring's + * TransactionSynchronization abstraction is used for something that is reading + * from the database, however, it is important because the same resultset is + * held open regardless of commits or roll backs. This means that when a + * transaction is committed, the input source is notified so that it can save + * it's current row number. Later, if the transaction is rolled back, the + * current row can be moved back to the same row number as it was on when commit + * was called. + *

+ * + *

+ * Calling skip will indicate to the input source that a record is bad and + * should not be represented to the user if the transaction is rolled back. For + * example, if row 2 is read in, and found to be bad, calling skip will inform + * the Input Source. If reading is then continued, and a rollback is necessary + * because of an error on output, the input source will be returned to row 1. + * Calling read while on row 1 will move the current row to 3, not 2, because 2 + * has been marked as skipped. + *

+ * + *

+ * Calling close on this Input Source will cause all resources it is currently + * using to be freed. (Connection, resultset, etc). If read() is called on the + * same instance again, the cursor will simply be reopened starting at row 0. + *

+ * + * @author Lucas Ward + * @author Peter Zozom + */ +public class SqlCursorInputSource implements InputSource, ResourceLifecycle, DisposableBean, + InitializingBean, Restartable, StatisticsProvider, Skippable { + + private static Log log = LogFactory.getLog(SqlCursorInputSource.class); + + public static final int VALUE_NOT_SET = -1; + + private static final String CURRENT_PROCESSED_ROW = "sqlCursorInput.lastProcessedRowNum"; + + private static final String SKIP_COUNT = "sqlCursorInput.skippedRrecordCount"; + + private Connection con; + + private Statement stmt; + + protected ResultSet rs; + + private DataSource dataSource; + + private String sql; + + private final List skippedRows = new ArrayList(); + + private int skipCount = 0; + + private int fetchSize = VALUE_NOT_SET; + + private int maxRows = VALUE_NOT_SET; + + private int queryTimeout = VALUE_NOT_SET; + + private boolean ignoreWarnings = true; + + private boolean verifyCursorPosition = true; + + private SQLExceptionTranslator exceptionTranslator; + + /* Current count of processed records. */ + private int currentProcessedRow = 0; + + private int lastCommittedRow = 0; + + private final SqlInputTransactionSynchronization transactionSynchronization = new SqlInputTransactionSynchronization(); + + private RowMapper mapper; + + private boolean initialized = false; + + /** + * Assert that mandatory properties are set. + * + * @throws IllegalArgumentException + * if either data source or sql properties not set. + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(dataSource, "DataSOurce must be provided"); + Assert.notNull(sql, "The SQL query must be provided"); + } + + /** + * Public setter for the data source for injection purposes. + * + * @param dataSource + */ + public void setDataSource(DataSource dataSource) { + this.dataSource = dataSource; + } + + /** + * Increment the cursor to the next row, validating the cursor position and + * passing the resultset to the RowMapper. If read has not been called on + * this instance before, the cursor will be opened. If there are skipped + * records for this commit scope, an internal list of skipped records will + * be checked to ensure that only a valid row is given to the mapper. + * + * @returns Object returned by RowMapper + * @throws DataAccessException + * @throws IllegalStateExceptino + * if mapper is null. + */ + public Object read() { + + if (!initialized) { + open(); + } + + Assert.state(mapper != null, "Mapper must not be null."); + + try { + if (!rs.next()) { + return null; + } else { + currentProcessedRow++; + if (!skippedRows.isEmpty()) { + // while is necessary to handle successive skips. + while (skippedRows + .contains(new Integer(currentProcessedRow))) { + if (!rs.next()) { + return null; + } + currentProcessedRow++; + } + } + + Object mappedResult = mapper.mapRow(rs, currentProcessedRow); + + verifyCursorPosition(currentProcessedRow); + + return mappedResult; + } + } catch (SQLException se) { + throw getExceptionTranslator().translate( + "Trying to process next row", sql, se); + } + + } + + public int getCurrentProcessedRow() { + return currentProcessedRow; + } + + /** + * Mark the current row. Calling reset will cause the result set to be set + * to the current row when mark was called. + */ + private void mark() { + lastCommittedRow = currentProcessedRow; + skippedRows.clear(); + } + + /** + * Set the ResultSet's current row to the last marked position. + * + * @throws DataAccessException + */ + private void reset() { + try { + currentProcessedRow = lastCommittedRow; + if (currentProcessedRow > 0) { + rs.absolute(currentProcessedRow); + } else { + rs.beforeFirst(); + } + + } catch (SQLException se) { + throw getExceptionTranslator().translate( + "Attempted to move ResultSet to last committed row", sql, + se); + } + } + + /** + * Close this input source. The ResultSet, Statement and Connection created + * will be closed. This must be called or the connection and cursor will be + * held open indefinitely! + * + * @see org.springframework.batch.item.ResourceLifecycle#close() + */ + public void close() { + initialized = false; + JdbcUtils.closeResultSet(this.rs); + JdbcUtils.closeStatement(this.stmt); + JdbcUtils.closeConnection(this.con); + this.currentProcessedRow = 0; + skippedRows.clear(); + skipCount = 0; + } + + /** + * Calls close to ensure that bean factories can close and always release + * resources. + * + * @see org.springframework.beans.factory.DisposableBean#destroy() + */ + public void destroy() throws Exception { + close(); + } + + // Check the result set is in synch with the currentRow attribute. This is + // important + // to ensure that the user hasn't modified the current row. + private void verifyCursorPosition(int expectedCurrentRow) + throws SQLException { + if (verifyCursorPosition) { + if (expectedCurrentRow != this.rs.getRow()) { + throw new InvalidDataAccessResourceUsageException( + "Unexpected cursor position change."); + } + } + } + + /* + * Executes the provided SQL query. The statement is created with + * 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' set to true. This is extremely + * important, since a non read-only cursor may lock tables that shouldn't be + * locked, and not holding the cursor open over a commit would require it to + * be reopened after each commit, which would destroy performance. + */ + private void executeQuery() { + + Assert.state(dataSource != null, "DataSource must not be null."); + + try { + this.con = dataSource.getConnection(); + this.stmt = this.con.createStatement( + ResultSet.TYPE_SCROLL_INSENSITIVE, + ResultSet.CONCUR_READ_ONLY, + ResultSet.HOLD_CURSORS_OVER_COMMIT); + applyStatementSettings(this.stmt); + this.rs = this.stmt.executeQuery(sql); + handleWarnings(this.stmt.getWarnings()); + } catch (SQLException se) { + close(); + throw getExceptionTranslator().translate("Executing query", + sql, se); + } + + BatchTransactionSynchronizationManager + .registerSynchronization(transactionSynchronization); + } + + /* + * Prepare the given JDBC Statement (or PreparedStatement or + * CallableStatement), applying statement settings such as fetch size, max + * rows, and query timeout. @param stmt the JDBC Statement to prepare + * @throws SQLException + * + * @see #setFetchSize + * @see #setMaxRows + * @see #setQueryTimeout + */ + private void applyStatementSettings(Statement stmt) throws SQLException { + if (fetchSize != VALUE_NOT_SET) { + stmt.setFetchSize(fetchSize); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + } + if (maxRows != VALUE_NOT_SET) { + stmt.setMaxRows(maxRows); + } + if (queryTimeout != VALUE_NOT_SET) { + stmt.setQueryTimeout(queryTimeout); + } + } + + /* + * Return the exception translator for this instance.

Creates a default + * SQLErrorCodeSQLExceptionTranslator for the specified DataSource if none + * is set. + */ + protected SQLExceptionTranslator getExceptionTranslator() { + if (exceptionTranslator == null) { + if (dataSource != null) { + exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator( + dataSource); + } else { + exceptionTranslator = new SQLStateSQLExceptionTranslator(); + } + } + return exceptionTranslator; + } + + /* + * Throw a SQLWarningException if we're not ignoring warnings, else log the + * warnings (at debug level). + * + * @param warning the warnings object from the current statement. May be + * null, in which case this method does nothing. + * + * @see org.springframework.jdbc.SQLWarningException + */ + private void handleWarnings(SQLWarning warnings) throws SQLWarningException { + if (ignoreWarnings) { + SQLWarning warningToLog = warnings; + while (warningToLog != null) { + log.debug("SQLWarning ignored: SQL state '" + + warningToLog.getSQLState() + "', error code '" + + warningToLog.getErrorCode() + "', message [" + + warningToLog.getMessage() + "]"); + warningToLog = warningToLog.getNextWarning(); + } + } else if(warnings != null){ + throw new SQLWarningException("Warning not ignored", warnings); + } + } + + /* + * (non-Javadoc) + * + * @see org.springframework.batch.restart.Restartable#getRestartData() + */ + public RestartData getRestartData() { + return new GenericRestartData(getStatistics()); + } + + /* + * (non-Javadoc) + * + * @see org.springframework.batch.restart.Restartable#restoreFrom(org.springframework.batch.restart.RestartData) + */ + public void restoreFrom(RestartData data) { + Assert.state(!initialized); + + if (data == null) + return; + + open(); + + Properties restartProperties = data.getProperties(); + if(restartProperties.containsKey(CURRENT_PROCESSED_ROW) == false){ + return; + } + + try { + this.currentProcessedRow = Integer.parseInt(restartProperties + .getProperty(CURRENT_PROCESSED_ROW)); + rs.absolute(currentProcessedRow); + } catch (SQLException se) { + throw getExceptionTranslator().translate( + "Attempted to move ResultSet to last committed row", sql, + se); + } + } + + /* + * (non-Javadoc) + * + * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() + */ + public Properties getStatistics() { + + Properties props = new Properties(); + props.setProperty(CURRENT_PROCESSED_ROW, new Integer( + currentProcessedRow).toString()); + props.setProperty(SKIP_COUNT, new Integer(skipCount).toString()); + return props; + } + + /** + * Skip the current row. If the transaction is rolled back, this row will + * not be represented to the RowMapper when read() is called. For example, + * if you read in row 2, find the data to be bad, and call skip(), then + * continue processing and find + */ + public void skip() { + skippedRows.add(new Integer(currentProcessedRow)); + skipCount++; + } + + /** + * Gives the JDBC driver a hint as to the number of rows that should be + * fetched from the database when more rows are needed for this + * ResultSet object. If the fetch size specified is zero, the + * JDBC driver ignores the value. + * + * @param fetchSize + * the number of rows to fetch + * @see ResultSet#setFetchSize(int) + */ + public void setFetchSize(int fetchSize) { + this.fetchSize = fetchSize; + } + + /** + * Sets the limit for the maximum number of rows that any + * ResultSet object can contain to the given number. + * + * @param maxRows + * the new max rows limit; zero means there is no limit + * @see Statement#setMaxRows(int) + */ + public void setMaxRows(int maxRows) { + this.maxRows = maxRows; + } + + /** + * Sets the number of seconds the driver will wait for a + * Statement object to execute to the given number of + * seconds. If the limit is exceeded, an SQLException is + * thrown. + * + * @param queryTimeout + * seconds the new query timeout limit in seconds; zero means + * there is no limit + * @see Statement#setQueryTimeout(int) + */ + public void setQueryTimeout(int queryTimeout) { + this.queryTimeout = queryTimeout; + } + + /** + * Set whether SQLWarnings should be ignored (only logged) or exception + * should be thrown. + * + * @param ignoreWarnings + * if TRUE, warnings are ignored + */ + public void setIgnoreWarnings(boolean ignoreWarnings) { + this.ignoreWarnings = ignoreWarnings; + } + + /** + * Allow verification of cursor position after current row is processed by + * RowMapper or RowCallbackHandler. Default value is TRUE. + * + * @param verifyCursorPosition + * if true, cursor position is verified + */ + public void setVerifyCursorPosition(boolean verifyCursorPosition) { + this.verifyCursorPosition = verifyCursorPosition; + } + + /** + * Set the RowMapper to be used for all calls to read(). + * + * @param mapper + */ + public void setMapper(RowMapper mapper) { + this.mapper = mapper; + } + + /** + * Set the sql statement to be used when creating the cursor. This statement + * should be a complete and valid Sql statement, as it will be run directly + * without any modification. + * + * @param sql + */ + public void setSql(String sql) { + this.sql = sql; + } + + public void open() { + Assert.isNull(rs); + executeQuery(); + initialized = true; + + } + + private class SqlInputTransactionSynchronization extends TransactionSynchronizationAdapter { + + /* + * @param status transaction status + * + * @see org.springframework.transaction.support.TransactionSynchronization#afterCompletion(int) + */ + public void afterCompletion(int status) { + + if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + reset(); + } else if (status == TransactionSynchronization.STATUS_COMMITTED) { + mark(); + } + } + } + +} diff --git a/infrastructure/src/main/java/org/springframework/batch/io/support/AbstractDrivingQueryInputSource.java b/infrastructure/src/main/java/org/springframework/batch/io/support/AbstractDrivingQueryInputSource.java new file mode 100644 index 000000000..338001693 --- /dev/null +++ b/infrastructure/src/main/java/org/springframework/batch/io/support/AbstractDrivingQueryInputSource.java @@ -0,0 +1,194 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.support; + +import java.util.Iterator; +import java.util.List; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.item.ResourceLifecycle; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationAdapter; +import org.springframework.util.Assert; + +/** + *

Abstract base class for driving query input sources. Input Sources of + * this type use a 'driving query' to return back a list of keys. Upon each + * call to read, a new key is returned.

+ * + *

Mutability: Because this base class cannot guarantee that the keys returned + * by subclasses are immutable, care should be taken to not modify a key value. + * Doing so would cause issues if a rollback occurs. For example, if a call + * to read() is made, and the returned key is modified, a rollback will cause + * the next call to read() to return the same object that was originally returned, + * since there is no way to create a defensive copy, and re-querying the database + * for all the keys would be too resource intensive.

+ * + * + * @author Lucas Ward + * + */ +public abstract class AbstractDrivingQueryInputSource implements InputSource, ResourceLifecycle, + DisposableBean, Restartable, InitializingBean { + + private boolean initialized = false; + + private List keys; + + private Iterator keysIterator; + + private int currentIndex = 0; + + private int lastCommitIndex = 0; + + private TransactionSynchronization synchronization = + new DrivingQueryInputSourceTransactionSynchronization(); + + /** + * Return the next key in the List. If the InputSource has not been initialized yet, + * then {@link AbstractDrivingQueryInputSource.open()} will be called. + * + * @return next key in the list if not index is not at the last element, null otherwise. + */ + public Object read() { + if (!initialized) { + open(); + } + + if (keysIterator.hasNext()) { + currentIndex++; + return keysIterator.next(); + } + + return null; + } + + /** + * Get the current key. This method will return the same + * object returned by the last read() method. If the + * InputSource hasn't been initialized yet, then null will + * be returned. + * + * @return the current key. + */ + protected Object getCurrentKey(){ + if(initialized){ + return keys.get(currentIndex - 1); + } + + return null; + } + + /** + * Close the resource by setting the list of keys to null, allowing them + * to be garbage collected. + */ + public void close() { + initialized = false; + currentIndex = 0; + lastCommitIndex = 0; + keys = null; + keysIterator = null; + } + + /** + * Initialize the input source by delegating to the subclass in order to retrieve + * the keys. The input source will also be registered with the + * {@link BatchTransactionSynchronizationManager} in order to ensure it is notified + * about commits and rollbacks. + * + * @throws IllegalStateException if the keys list is null or initialized is true. + */ + public void open() { + + Assert.state(keys == null || initialized, "Cannot open an already opened input source" + + ", call close() first."); + keys = retrieveKeys(); + keysIterator = keys.listIterator(); + BatchTransactionSynchronizationManager.registerSynchronization(synchronization); + initialized = true; + } + + /* (non-Javadoc) + * @see org.springframework.beans.factory.DisposableBean#destroy() + */ + public void destroy() throws Exception { + close(); + } + + /** + * Restore input source to previous state. If the input source has already + * been initialized before calling restore (meaning, read has been called) + * then an IllegalStateException will be thrown, since all input sources + * should be restored before being read from, otherwise already processed + * data could be returned. The RestartData attempting to be restored from + * must have been obtained from the same input source as the one + * being restored from otherwise it is invalid. + * + * @throws IllegalArgumentException if restart data or it's properties is null. + * @throws IllegalStateException if the input source has already been intialized. + */ + public void restoreFrom(RestartData data) { + + Assert.notNull(data, "RestartData must not be null."); + Assert.notNull(data.getProperties(), "RestartData properties must not be null."); + Assert.state(!initialized, "Cannot restore when already intialized. Call" + + " close() first before restore()"); + + if (data.getProperties().size() == 0) { + return; + } + + keys = restoreKeys(data); + + if(keys != null & keys.size() > 0){ + keysIterator = keys.listIterator(); + initialized = true; + } + } + + //Abstract Methods + /** + * @return list of keys returned by the driving query + */ + protected abstract List retrieveKeys(); + + /** + * Restore the keys list based on provided restart data. + * + * @param restartData, the restart data to restore the keys list from. + * @return a list of keys. + */ + protected abstract List restoreKeys(RestartData restartData); + + /** + * Encapsulates transaction events handling. + */ + private class DrivingQueryInputSourceTransactionSynchronization extends TransactionSynchronizationAdapter { + public void afterCompletion(int status) { + if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + keysIterator = keys.listIterator(lastCommitIndex); + } else if (status == TransactionSynchronization.STATUS_COMMITTED) { + lastCommitIndex = currentIndex; + } + } + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/orm/hibernate/HibernateInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/orm/hibernate/HibernateInputSourceIntegrationTests.java new file mode 100644 index 000000000..16c4e5a2c --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/orm/hibernate/HibernateInputSourceIntegrationTests.java @@ -0,0 +1,36 @@ +package org.springframework.batch.io.orm.hibernate; + +import org.hibernate.SessionFactory; +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.orm.hibernate.HibernateInputSource; +import org.springframework.batch.io.support.AbstractDataSourceInputSourceIntegrationTests; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import org.springframework.orm.hibernate3.LocalSessionFactoryBean; + +/** + * Tests for {@link HibernateInputSource} + * + * @author Robert Kasanicky + */ +public class HibernateInputSourceIntegrationTests extends AbstractDataSourceInputSourceIntegrationTests { + + protected InputSource createInputSource() throws Exception { + LocalSessionFactoryBean factoryBean = new LocalSessionFactoryBean(); + factoryBean.setDataSource(super.getJdbcTemplate().getDataSource()); + factoryBean.setMappingLocations(new Resource[]{new ClassPathResource("Foo.hbm.xml", getClass())}); + factoryBean.afterPropertiesSet(); + + SessionFactory sessionFactory = (SessionFactory) factoryBean.getObject(); + + String hsqlQuery = "from Foo"; + + HibernateInputSource inputSource = new HibernateInputSource(); + inputSource.setQueryString(hsqlQuery); + inputSource.setSessionFactory(sessionFactory); + inputSource.afterPropertiesSet(); + + return inputSource; + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/orm/ibatis/IbatisInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/orm/ibatis/IbatisInputSourceIntegrationTests.java new file mode 100644 index 000000000..c72041804 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/orm/ibatis/IbatisInputSourceIntegrationTests.java @@ -0,0 +1,36 @@ +package org.springframework.batch.io.orm.ibatis; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.orm.ibatis.IbatisDrivingQueryInputSource; +import org.springframework.batch.io.support.AbstractDataSourceInputSourceIntegrationTests; +import org.springframework.core.io.ClassPathResource; +import org.springframework.orm.ibatis.SqlMapClientFactoryBean; + +import com.ibatis.sqlmap.client.SqlMapClient; + +/** + * Tests for {@link IbatisDrivingQueryInputSource} + * + * @author Robert Kasanicky + */ +public class IbatisInputSourceIntegrationTests extends AbstractDataSourceInputSourceIntegrationTests { + + protected InputSource createInputSource() throws Exception { + + SqlMapClientFactoryBean factory = new SqlMapClientFactoryBean(); + factory.setConfigLocation(new ClassPathResource("ibatis-config.xml", getClass())); + factory.setDataSource(super.getJdbcTemplate().getDataSource()); + factory.afterPropertiesSet(); + SqlMapClient sqlMapClient = (SqlMapClient) factory.getObject(); + + IbatisInputSource inputSource = new IbatisInputSource(); + inputSource.setDrivingQueryId("getAllFooIds"); + inputSource.setDetailsQueryId("getFooById"); + inputSource.setRestartQueryId("getAllFooIdsRestart"); + inputSource.setSqlMapClient(sqlMapClient); + + return inputSource; + } + + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sample/domain/Foo.java b/infrastructure/src/test/java/org/springframework/batch/io/sample/domain/Foo.java new file mode 100644 index 000000000..8c64986db --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sample/domain/Foo.java @@ -0,0 +1,54 @@ +package org.springframework.batch.io.sample.domain; + +import org.apache.commons.lang.builder.EqualsBuilder; +import org.apache.commons.lang.builder.HashCodeBuilder; + +/** + * Simple domain object for testing purposes. + */ +public class Foo { + + private int id; + private String name; + private int value; + + public Foo(){} + + public Foo(int id, String name, int value) { + this.id = id; + this.name = name; + this.value = value; + } + + public String getName() { + return name; + } + public void setName(String name) { + this.name = name; + } + public int getValue() { + return value; + } + public void setValue(int value) { + this.value = value; + } + public int getId() { + return id; + } + public void setId(int id) { + this.id = id; + } + + public String toString() { + return "Foo[id=" +id +",name=" + name + ",value=" + value + "]"; + } + + public boolean equals(Object obj) { + return EqualsBuilder.reflectionEquals(this, obj); + } + + public int hashCode() { + return HashCodeBuilder.reflectionHashCode(this); + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java new file mode 100644 index 000000000..26b1872bd --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java @@ -0,0 +1,183 @@ +package org.springframework.batch.io.sql; + +import java.util.Properties; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.restart.GenericRestartData; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionSynchronizationUtils; +import org.springframework.util.Assert; + +/** + * Common scenarios for testing {@link InputSource} implementations which read data from database. + * + * @author Lucas Ward + * @author Robert Kasanicky + */ +public abstract class AbstractSqlInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests { + + protected InputSource source; + + + /** + * @return input source with all necessary dependencies set + */ + protected abstract InputSource createInputSource() throws Exception; + + protected String[] getConfigLocations(){ + return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"}; + } + + protected void onSetUp()throws Exception{ + super.onSetUp(); + BatchTransactionSynchronizationManager.clearSynchronizations(); + source = createInputSource(); + getAsInitializingBean(source).afterPropertiesSet(); + } + + protected void onTearDown()throws Exception { + getAsDisposableBean(source).destroy(); + BatchTransactionSynchronizationManager.clearSynchronizations(); + super.onTearDown(); + } + + /** + * Regular scenario - read all rows and eventually return null. + */ + public void testNormalProcessing() throws Exception { + getAsInitializingBean(source).afterPropertiesSet(); + + Foo foo1 = (Foo) source.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) source.read(); + assertEquals(2, foo2.getValue()); + + Foo foo3 = (Foo) source.read(); + assertEquals(3, foo3.getValue()); + + Foo foo4 = (Foo) source.read(); + assertEquals(4, foo4.getValue()); + + Foo foo5 = (Foo) source.read(); + assertEquals(5, foo5.getValue()); + + assertNull(source.read()); + } + + /** + * Restart scenario. + * @throws Exception + */ + public void testRestart() throws Exception { + + Foo foo1 = (Foo) source.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) source.read(); + assertEquals(2, foo2.getValue()); + + RestartData restartData = getAsRestartable(source).getRestartData(); + + // create new input source + source = createInputSource(); + + getAsRestartable(source).restoreFrom(restartData); + + Foo fooAfterRestart = (Foo) source.read(); + assertEquals(3, fooAfterRestart.getValue()); + } + + /** + * Reading from an input source and then trying to restore causes an error. + */ + public void testInvalidRestore() throws Exception { + + Foo foo1 = (Foo) source.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) source.read(); + assertEquals(2, foo2.getValue()); + + RestartData restartData = getAsRestartable(source).getRestartData(); + + // create new input source + source = createInputSource(); + + Foo foo = (Foo) source.read(); + assertEquals(1, foo.getValue()); + + try { + getAsRestartable(source).restoreFrom(restartData); + fail(); + } + catch (IllegalStateException ex) { + // expected + } + } + + /** + * Empty restart data should be handled gracefully. + */ + public void testRestoreFromEmptyData() { + RestartData restartData = new GenericRestartData(new Properties()); + + getAsRestartable(source).restoreFrom(restartData); + + Foo foo = (Foo) source.read(); + assertEquals(1, foo.getValue()); + } + + /** + * Rollback scenario. + */ + public void testRollback() { + Foo foo1 = (Foo) source.read(); + + commit(); + + Foo foo2 = (Foo) source.read(); + Assert.state(!foo2.equals(foo1)); + + Foo foo3 = (Foo) source.read(); + Assert.state(!foo2.equals(foo3)); + + rollback(); + + assertEquals(foo2, source.read()); + } + + + private void commit() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_COMMITTED); + } + + private void rollback() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_ROLLED_BACK); + } + + private Restartable getAsRestartable(InputSource source) { + return (Restartable) source; + } + + private InitializingBean getAsInitializingBean(InputSource source) { + return (InitializingBean) source; + } + + private DisposableBean getAsDisposableBean(InputSource source) { + return (DisposableBean) source; + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeyFooDao.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeyFooDao.java new file mode 100644 index 000000000..580fce740 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeyFooDao.java @@ -0,0 +1,58 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.sql; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.List; + +import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.support.JdbcDaoSupport; + +/** + * @author Lucas Ward + * + */ +public class CompositeKeyFooDao extends JdbcDaoSupport implements FooDao { + + public CompositeKeyFooDao(JdbcTemplate jdbcTemplate) { + this.setJdbcTemplate(jdbcTemplate); + } + /* (non-Javadoc) + * @see org.springframework.batch.io.sql.scratch.FooDao#getFoo(java.lang.Object) + */ + public Foo getFoo(Object key) { + + List keys = (List)key; + Object[] args = new Object[]{keys.get(0), keys.get(1)}; + + RowMapper fooMapper = new RowMapper(){ + public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + Foo foo = new Foo(); + foo.setId(rs.getInt(1)); + foo.setName(rs.getString(2)); + foo.setValue(rs.getInt(3)); + return foo; + } + }; + + return (Foo)getJdbcTemplate().query("SELECT ID, NAME, VALUE from T_FOOS where ID = ? and VALUE = ?", + args, fooMapper).get(0); + } + +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSourceIntegrationTests.java new file mode 100644 index 000000000..5a1c88566 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSourceIntegrationTests.java @@ -0,0 +1,81 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.sql; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.sql.CompositeKeySqlDrivingQueryInputSource; +import org.springframework.batch.restart.GenericRestartData; +import org.springframework.batch.restart.RestartData; +import org.springframework.jdbc.core.RowMapper; + +/** + * @author Lucas Ward + * + */ +public class CompositeKeySqlDrivingQueryInputSourceIntegrationTests extends + AbstractSqlInputSourceIntegrationTests { + + protected InputSource createInputSource() throws Exception { + + CompositeKeySqlDrivingQueryInputSource inputSource = + new CompositeKeySqlDrivingQueryInputSource(getJdbcTemplate(), + "SELECT ID, VALUE from T_FOOS order by ID, VALUE", + new FooCompositeKeyMapper()); + + inputSource.setRestartQuery("SELECT ID from T_FOOS where ID > ? and VALUE > ? order by ID"); + inputSource.setRestartDataConverter(new FooRestartDataConverter()); + FooInputSource fooInputSource = new FooInputSource(inputSource, getJdbcTemplate()); + fooInputSource.setFooDao(new CompositeKeyFooDao(getJdbcTemplate())); + return fooInputSource; + } + + private class FooRestartDataConverter implements CompositeKeyRestartDataConverter{ + + private static final String ID_RESTART_KEY = "FooRestartDataConverter.id"; + private static final String VALUE_RESTART_KEY = "FooRestartDataConverter.value"; + + public RestartData createRestartData(Object compositeKey) { + + List values = (List)compositeKey; + Properties data = new Properties(); + data.setProperty(ID_RESTART_KEY, values.get(0).toString()); + data.setProperty(VALUE_RESTART_KEY, values.get(1).toString()); + return new GenericRestartData(data); + } + + public Object[] createArguments(RestartData restartData) { + Object[] args = new Object[2]; + args[0] = restartData.getProperties().get(ID_RESTART_KEY); + args[1] = restartData.getProperties().getProperty(VALUE_RESTART_KEY); + return args; + } + } + + private class FooCompositeKeyMapper implements RowMapper{ + public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + List key = new ArrayList(); + key.add(new Long(rs.getLong(1))); + key.add(new Long(rs.getLong(2))); + return key; + } + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/FooDao.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/FooDao.java new file mode 100644 index 000000000..7c258544b --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/FooDao.java @@ -0,0 +1,30 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.io.sql; + +import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.jdbc.core.JdbcTemplate; + +/** + * @author Lucas Ward + * + */ +public interface FooDao { + + Foo getFoo(Object key); + + void setJdbcTemplate(JdbcTemplate jdbcTemplate); +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/FooInputSource.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/FooInputSource.java new file mode 100644 index 000000000..a273960f0 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/FooInputSource.java @@ -0,0 +1,48 @@ +package org.springframework.batch.io.sql; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.support.AbstractDrivingQueryInputSource; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.jdbc.core.JdbcTemplate; + +class FooInputSource implements InputSource, Restartable, DisposableBean, InitializingBean{ + + AbstractDrivingQueryInputSource inputSource; + FooDao fooDao = new SingleKeyFooDao(); + + public FooInputSource(AbstractDrivingQueryInputSource inputSource, JdbcTemplate jdbcTemplate) { + this.inputSource = inputSource; + fooDao.setJdbcTemplate(jdbcTemplate); + } + + public Object read() { + Object key = inputSource.read(); + if(key != null){ + return fooDao.getFoo(key); + }else{ + return null; + } + } + + public RestartData getRestartData() { + return inputSource.getRestartData(); + } + + public void restoreFrom(RestartData data) { + inputSource.restoreFrom(data); + } + + public void destroy() throws Exception { + inputSource.destroy(); + } + + public void setFooDao(FooDao fooDao) { + this.fooDao = fooDao; + } + + public void afterPropertiesSet() throws Exception { + }; +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/FooRowMapper.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/FooRowMapper.java new file mode 100644 index 000000000..f7c038b82 --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/FooRowMapper.java @@ -0,0 +1,21 @@ +package org.springframework.batch.io.sql; + +import java.sql.ResultSet; +import java.sql.SQLException; + +import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.jdbc.core.RowMapper; + + +public class FooRowMapper implements RowMapper { + + public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + + Foo foo = new Foo(); + foo.setId(rs.getInt(1)); + foo.setName(rs.getString(2)); + foo.setValue(rs.getInt(3)); + + return foo; + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeyFooDao.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeyFooDao.java new file mode 100644 index 000000000..4e5866bfb --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeyFooDao.java @@ -0,0 +1,28 @@ +package org.springframework.batch.io.sql; + +import java.sql.ResultSet; +import java.sql.SQLException; + +import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.support.JdbcDaoSupport; + +public class SingleKeyFooDao extends JdbcDaoSupport implements FooDao { + + public Foo getFoo(Object key){ + + RowMapper fooMapper = new RowMapper(){ + public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + Foo foo = new Foo(); + foo.setId(rs.getInt(1)); + foo.setName(rs.getString(2)); + foo.setValue(rs.getInt(3)); + return foo; + } + }; + + return (Foo)getJdbcTemplate().query("SELECT ID, NAME, VALUE from T_FOOS where ID = ?", + new Object[] {key}, fooMapper).get(0); + + } +} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSourceIntegrationTests.java deleted file mode 100644 index 97c70f6b5..000000000 --- a/infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSourceIntegrationTests.java +++ /dev/null @@ -1,152 +0,0 @@ -package org.springframework.batch.io.sql; - -import java.sql.ResultSet; -import java.sql.SQLException; - -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.restart.RestartData; -import org.springframework.jdbc.core.RowMapper; -import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; - -public class SingleKeySqlDrivingQueryInputSourceIntegrationTests - extends AbstractTransactionalDataSourceSpringContextTests { - - SingleKeySqlDrivingQueryInputSource sqlInputSource; - - protected String[] getConfigLocations(){ - return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"}; - } - - protected void onSetUp()throws Exception{ - super.onSetUp(); - - sqlInputSource = createInputSource(); - } - - protected SingleKeySqlDrivingQueryInputSource createInputSource(){ - - SingleKeySqlDrivingQueryInputSource inputSource = new SingleKeySqlDrivingQueryInputSource(); - inputSource.setDrivingQuery("SELECT ID from T_FOOS order by ID"); - inputSource.setDetailsQuery("SELECT NAME, VALUE from T_FOOS where ID = ?"); - inputSource.setRestartQuery("SELECT ID from T_FOOS where ID > ? order by ID"); - inputSource.setMapper(new FooMapper()); - inputSource.setDataSource(super.getJdbcTemplate().getDataSource()); - return inputSource; - } - - protected void onTearDown()throws Exception{ - - BatchTransactionSynchronizationManager.clearSynchronizations(); - sqlInputSource.close(); - super.onTearDown(); - } - - public void testNormalProcessing(){ - - Foo foo = (Foo)sqlInputSource.read(); - assertEquals(1, foo.value); - - foo = (Foo)sqlInputSource.read(); - assertEquals(2, foo.value); - - foo = (Foo)sqlInputSource.read(); - assertEquals(3, foo.value); - - foo = (Foo)sqlInputSource.read(); - assertEquals(4, foo.value); - - foo = (Foo)sqlInputSource.read(); - assertEquals(5, foo.value); - - assertNull(sqlInputSource.read()); - } - -/* public void testRollback(){ - - SqlIdentityKey key = sqlInputSource.readKey(); - assertEquals("1", key.getKeyValue("id")); - - key = sqlInputSource.readKey(); - assertEquals("2", key.getKeyValue("id")); - - super.setComplete(); - super.endTransaction(); - super.startNewTransaction(); - BatchTransactionSynchronizationManager.resynchronize(); - - key = sqlInputSource.readKey(); - assertEquals("3", key.getKeyValue("id")); - - key = sqlInputSource.readKey(); - assertEquals("4", key.getKeyValue("id")); - - super.endTransaction(); - super.startNewTransaction(); - - key = sqlInputSource.readKey(); - assertEquals("3", key.getKeyValue("id")); - - }*/ - - public void testRestart(){ - - Foo foo = (Foo)sqlInputSource.read(); - assertEquals(1, foo.value); - - foo = (Foo)sqlInputSource.read(); - assertEquals(2, foo.value); - - RestartData restartData = sqlInputSource.getRestartData(); - - //create new input source - sqlInputSource = createInputSource(); - - sqlInputSource.restoreFrom(restartData); - - foo = (Foo)sqlInputSource.read(); - assertEquals(3, foo.value); - } - - //test that reading from an input source and then trying to restore causes an error. - public void testInvalidRestore(){ - - Foo foo = (Foo)sqlInputSource.read(); - assertEquals(1, foo.value); - - foo = (Foo)sqlInputSource.read(); - assertEquals(2, foo.value); - - RestartData restartData = sqlInputSource.getRestartData(); - - //create new input source - sqlInputSource = createInputSource(); - - foo = (Foo)sqlInputSource.read(); - assertEquals(1, foo.value); - - try{ - sqlInputSource.restoreFrom(restartData); - fail(); - } - catch(IllegalStateException ex){ - //expected - } - } - - private class Foo { - String name; - int value; - } - - private class FooMapper implements RowMapper{ - - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - - Foo foo = new Foo(); - foo.name = rs.getString(1); - foo.value = rs.getInt(2); - return foo; - } - - } -} diff --git a/infrastructure/src/test/java/org/springframework/batch/io/sql/SqlCursorInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/sql/SqlCursorInputSourceIntegrationTests.java index 1de768539..d4446e607 100644 --- a/infrastructure/src/test/java/org/springframework/batch/io/sql/SqlCursorInputSourceIntegrationTests.java +++ b/infrastructure/src/test/java/org/springframework/batch/io/sql/SqlCursorInputSourceIntegrationTests.java @@ -1,336 +1,27 @@ package org.springframework.batch.io.sql; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.util.Properties; - -import org.springframework.batch.repeat.context.RepeatContextSupport; -import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; -import org.springframework.batch.restart.GenericRestartData; -import org.springframework.batch.restart.RestartData; -import org.springframework.dao.DataAccessException; -import org.springframework.dao.InvalidDataAccessResourceUsageException; -import org.springframework.jdbc.core.RowMapper; -import org.springframework.jdbc.datasource.SingleConnectionDataSource; -import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; +import org.springframework.batch.io.InputSource; /** - * - * @author Lucas Ward - * + * Tests for {@link SqlCursorInputSource} + * + * @author Robert Kasanicky */ -public class SqlCursorInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests { - - protected SqlCursorInputSource sqlCursorInputSource; - - private RowMapper mapper = new SqlRowMapper(); - - private static final String CURRENT_PROCESSED_ROW = "sqlCursorInput.lastProcessedRowNum"; - private static final String SKIP_COUNT = "sqlCursorInput.skippedRrecordCount"; - - protected String[] getConfigLocations(){ - return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"}; - } - - protected void onSetUp()throws Exception{ - sqlCursorInputSource = getNewInputSource(); - sqlCursorInputSource.setMapper(mapper); - sqlCursorInputSource.setFetchSize(10); - sqlCursorInputSource.setMaxRows(100); - sqlCursorInputSource.setQueryTimeout(1000); - RepeatSynchronizationManager.register(new RepeatContextSupport(null)); - super.onSetUp(); - } - - protected void onTearDown()throws Exception{ - //cursor must be closed between each test, and transaction synchronization - //list must be cleared. - BatchTransactionSynchronizationManager.clearSynchronizations(); - RepeatSynchronizationManager.clear(); - sqlCursorInputSource.destroy(); - super.onTearDown(); - } - - protected SqlCursorInputSource getNewInputSource(){ +public class SqlCursorInputSourceIntegrationTests extends AbstractSqlInputSourceIntegrationTests{ + protected InputSource createInputSource() throws Exception { SqlCursorInputSource result = new SqlCursorInputSource(); result.setDataSource(super.getJdbcTemplate().getDataSource()); - result.setSql("SELECT * from T_FOOS"); + result.setSql("select ID, NAME, VALUE from T_FOOS"); result.setIgnoreWarnings(true); result.setVerifyCursorPosition(true); + + result.setMapper(new FooRowMapper()); + result.setFetchSize(10); + result.setMaxRows(100); + result.setQueryTimeout(1000); return result; } - public void testAfterPropertiesSet() throws Exception{ - //all dependencies have been set, so there shouldn't be any exceptions - sqlCursorInputSource.afterPropertiesSet(); - - sqlCursorInputSource.setSql(null); - - try{ - sqlCursorInputSource.afterPropertiesSet(); - fail(); - }catch(IllegalArgumentException ex){ - //expected - } - } - - public void testNormalReading(){ - - int fooCount = 0; - - for(;;){ - Foo foo = (Foo)sqlCursorInputSource.read(); - - if( foo == null){ - break; - } - - fooCount++; - validateFoo(fooCount, "bar" + fooCount, fooCount, foo); - assertEquals(sqlCursorInputSource.getCurrentProcessedRow(), fooCount); - - } - - assertEquals(5, fooCount ); - } - - public void testModifyCursorPosition(){ - - sqlCursorInputSource.setMapper(new RowMapper(){ - - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - rs.next(); - return null; - }}); - - try{ - sqlCursorInputSource.read(); - fail(); - }catch(InvalidDataAccessResourceUsageException ex){ - //expected - } - } - - - public void testRestart(){ - - sqlCursorInputSource.read(); - Foo foo = (Foo)sqlCursorInputSource.read(); - - validateFoo(2, "bar2", 2, foo); - - RestartData restartData = sqlCursorInputSource.getRestartData(); - - sqlCursorInputSource = getNewInputSource(); - sqlCursorInputSource.setMapper(mapper); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(1, "bar1", 1, foo); - - sqlCursorInputSource.restoreFrom(restartData); - - foo = (Foo)sqlCursorInputSource.read(); - - validateFoo(3, "bar3", 3, foo); - } - - public void testRestartWithEmptyRestartData(){ - - RestartData restartData = new GenericRestartData(new Properties()); - sqlCursorInputSource.restoreFrom(restartData); - } - - public void testReadWithNullMapper(){ - - //calling read without a mapper should throw an exception. - sqlCursorInputSource.setMapper(null); - try{ - sqlCursorInputSource.read(); - fail(); - } - catch(IllegalStateException ex){ - //expected - } - } - - public void testStatistics(){ - - Properties statistics = sqlCursorInputSource.getStatistics(); - assertEquals("0", statistics.getProperty(CURRENT_PROCESSED_ROW)); - sqlCursorInputSource.read(); - - statistics = sqlCursorInputSource.getStatistics(); - assertEquals("1", statistics.getProperty(CURRENT_PROCESSED_ROW)); - } - - public void testSkipCountStatistics(){ - - sqlCursorInputSource.read(); - Foo foo = (Foo)sqlCursorInputSource.read(); - validateFoo(2, "bar2", 2, foo); - - Properties statistics = sqlCursorInputSource.getStatistics(); - assertEquals("0", statistics.getProperty(SKIP_COUNT)); - - sqlCursorInputSource.skip(); - - statistics = sqlCursorInputSource.getStatistics(); - assertEquals("1", statistics.getProperty(SKIP_COUNT)); - - sqlCursorInputSource.read(); - - sqlCursorInputSource.read(); - sqlCursorInputSource.skip(); - - statistics = sqlCursorInputSource.getStatistics(); - assertEquals("2", statistics.getProperty(SKIP_COUNT)); - - super.endTransaction(); - super.startNewTransaction(); - - sqlCursorInputSource.read(); - - statistics = sqlCursorInputSource.getStatistics(); - assertEquals("2", statistics.getProperty(SKIP_COUNT)); - } - - public void testRollback(){ - - sqlCursorInputSource.read(); - Foo foo = (Foo)sqlCursorInputSource.read(); - validateFoo(2, "bar2", 2, foo); - - super.setComplete(); - super.endTransaction(); - super.startNewTransaction(); - BatchTransactionSynchronizationManager.resynchronize(); - - sqlCursorInputSource.read(); - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(4, "bar4", 4, foo); - - super.endTransaction(); - super.startNewTransaction(); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(3, "bar3", 3, foo); - } - - public void testSkip(){ - - sqlCursorInputSource.read(); - Foo foo = (Foo)sqlCursorInputSource.read(); - validateFoo(2, "bar2", 2, foo); - - sqlCursorInputSource.skip(); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(3, "bar3", 3, foo); - - super.endTransaction(); - super.startNewTransaction(); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(1, "bar1", 1, foo); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(3, "bar3", 3, foo); - } - - public void testSucessiveSkip(){ - - sqlCursorInputSource.read(); - Foo foo = (Foo)sqlCursorInputSource.read(); - validateFoo(2, "bar2", 2, foo); - sqlCursorInputSource.skip(); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(3, "bar3", 3, foo); - sqlCursorInputSource.skip(); - - super.endTransaction(); - super.startNewTransaction(); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(1, "bar1", 1, foo); - - foo = (Foo)sqlCursorInputSource.read(); - validateFoo(4, "bar4", 4, foo); - } - - public void testMappingException(){ - - sqlCursorInputSource.setMapper(new RowMapper(){ - - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - throw new SQLException(); - }}); - - try{ - sqlCursorInputSource.read(); - fail(); - }catch(DataAccessException ex){ - //expected - } - } - - public void testExecuteQueryException(){ - - sqlCursorInputSource.setDataSource(new ExceptionDataSource()); - - try{ - sqlCursorInputSource.read(); - fail(); - }catch(DataAccessException ex){ - //expected - } - } - - public void testNoIgnoreWarnings(){ - - //there shouldn't be any exceptions if ignore warnings is false - //and there are no warnings. - sqlCursorInputSource.setIgnoreWarnings(false); - sqlCursorInputSource.read(); - } - - - private void validateFoo(int id, String name, int value, Foo foo){ - - assertEquals(id, foo.id); - assertEquals(name, foo.name); - assertEquals(value, foo.value); - } - - private class SqlRowMapper implements RowMapper { - - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - - Foo foo = new Foo(); - foo.id = rs.getInt(1); - foo.name = rs.getString(2); - foo.value = rs.getInt(3); - - return foo; - } - - } - - private class Foo{ - - private int id; - private String name; - private int value; - } - - private class ExceptionDataSource extends SingleConnectionDataSource{ - - public Connection getConnection() throws SQLException { - throw new SQLException(); - } - } } diff --git a/infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceInputSourceIntegrationTests.java b/infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceInputSourceIntegrationTests.java new file mode 100644 index 000000000..ac65841ab --- /dev/null +++ b/infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceInputSourceIntegrationTests.java @@ -0,0 +1,183 @@ +package org.springframework.batch.io.support; + +import java.util.Properties; + +import org.springframework.batch.io.InputSource; +import org.springframework.batch.io.sample.domain.Foo; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.restart.GenericRestartData; +import org.springframework.batch.restart.RestartData; +import org.springframework.batch.restart.Restartable; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionSynchronizationUtils; +import org.springframework.util.Assert; + +/** + * Common scenarios for testing {@link InputSource} implementations which read data from database. + * + * @author Lucas Ward + * @author Robert Kasanicky + */ +public abstract class AbstractDataSourceInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests { + + protected InputSource source; + + + /** + * @return configured input source ready for use + */ + protected abstract InputSource createInputSource() throws Exception; + + protected String[] getConfigLocations(){ + return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"}; + } + + protected void onSetUp()throws Exception{ + super.onSetUp(); + BatchTransactionSynchronizationManager.clearSynchronizations(); + source = createInputSource(); + } + + protected void onTearDown()throws Exception { + getAsDisposableBean(source).destroy(); + BatchTransactionSynchronizationManager.clearSynchronizations(); + super.onTearDown(); + } + + /** + * Regular scenario - read all rows and eventually return null. + */ + public void testNormalProcessing() throws Exception { + getAsInitializingBean(source).afterPropertiesSet(); + + Foo foo1 = (Foo) source.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) source.read(); + assertEquals(2, foo2.getValue()); + + Foo foo3 = (Foo) source.read(); + assertEquals(3, foo3.getValue()); + + Foo foo4 = (Foo) source.read(); + assertEquals(4, foo4.getValue()); + + Foo foo5 = (Foo) source.read(); + assertEquals(5, foo5.getValue()); + + assertNull(source.read()); + } + + /** + * Restart scenario - read records, save restart data, create new input source + * and restore from restart data - the new input source should continue where + * the old one finished. + */ + public void testRestart() throws Exception { + + Foo foo1 = (Foo) source.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) source.read(); + assertEquals(2, foo2.getValue()); + + RestartData restartData = getAsRestartable(source).getRestartData(); + + // create new input source + source = createInputSource(); + + getAsRestartable(source).restoreFrom(restartData); + + Foo fooAfterRestart = (Foo) source.read(); + assertEquals(3, fooAfterRestart.getValue()); + } + + /** + * Reading from an input source and then trying to restore causes an error. + */ + public void testInvalidRestore() throws Exception { + + Foo foo1 = (Foo) source.read(); + assertEquals(1, foo1.getValue()); + + Foo foo2 = (Foo) source.read(); + assertEquals(2, foo2.getValue()); + + RestartData restartData = getAsRestartable(source).getRestartData(); + + // create new input source + source = createInputSource(); + + Foo foo = (Foo) source.read(); + assertEquals(1, foo.getValue()); + + try { + getAsRestartable(source).restoreFrom(restartData); + fail(); + } + catch (IllegalStateException ex) { + // expected + } + } + + /** + * Empty restart data should be handled gracefully. + */ + public void testRestoreFromEmptyData() { + RestartData restartData = new GenericRestartData(new Properties()); + + getAsRestartable(source).restoreFrom(restartData); + + Foo foo = (Foo) source.read(); + assertEquals(1, foo.getValue()); + } + + /** + * Rollback scenario - input source rollbacks to last commit point. + */ + public void testRollback() { + Foo foo1 = (Foo) source.read(); + + commit(); + + Foo foo2 = (Foo) source.read(); + Assert.state(!foo2.equals(foo1)); + + Foo foo3 = (Foo) source.read(); + Assert.state(!foo2.equals(foo3)); + + rollback(); + + assertEquals(foo2, source.read()); + } + + + private void commit() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_COMMITTED); + } + + private void rollback() { + TransactionSynchronizationUtils.invokeAfterCompletion( + TransactionSynchronizationManager.getSynchronizations(), + TransactionSynchronization.STATUS_ROLLED_BACK); + } + + private Restartable getAsRestartable(InputSource source) { + return (Restartable) source; + } + + private InitializingBean getAsInitializingBean(InputSource source) { + return (InitializingBean) source; + } + + private DisposableBean getAsDisposableBean(InputSource source) { + return (DisposableBean) source; + } + +} diff --git a/infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo-write.hbm.xml b/infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo-write.hbm.xml new file mode 100644 index 000000000..67b35901c --- /dev/null +++ b/infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo-write.hbm.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo.hbm.xml b/infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo.hbm.xml new file mode 100644 index 000000000..8a23dece9 --- /dev/null +++ b/infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo.hbm.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-config.xml b/infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-config.xml new file mode 100644 index 000000000..3581245fd --- /dev/null +++ b/infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-config.xml @@ -0,0 +1,12 @@ + + + + + + + + + + \ No newline at end of file diff --git a/infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-foo.xml b/infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-foo.xml new file mode 100644 index 000000000..bed651148 --- /dev/null +++ b/infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-foo.xml @@ -0,0 +1,41 @@ + + + + + + + + + + + + + + + + + + + + + insert INTO T_WRITE_FOOS (ID, NAME, VALUE) VALUES (#id#, #name#, #value#) + + + + update T_WRITE_FOOS set NAME = #name#, VALUE = #value# where ID = #id# + + + + delete from T_WRITE_FOOS where ID = #id# + + + \ No newline at end of file diff --git a/infrastructure/src/test/resources/org/springframework/batch/io/sql/destroy-foo-schema-hsqldb.sql b/infrastructure/src/test/resources/org/springframework/batch/io/sql/destroy-foo-schema-hsqldb.sql index 7124ced3e..34ac26db0 100644 --- a/infrastructure/src/test/resources/org/springframework/batch/io/sql/destroy-foo-schema-hsqldb.sql +++ b/infrastructure/src/test/resources/org/springframework/batch/io/sql/destroy-foo-schema-hsqldb.sql @@ -1 +1,2 @@ -DROP TABLE T_FOOS; \ No newline at end of file +DROP TABLE T_FOOS; +DROP TABLE T_WRITE_FOOS; \ No newline at end of file diff --git a/infrastructure/src/test/resources/org/springframework/batch/io/sql/init-foo-schema-hsqldb.sql b/infrastructure/src/test/resources/org/springframework/batch/io/sql/init-foo-schema-hsqldb.sql index 2890a4823..c6b0e562f 100644 --- a/infrastructure/src/test/resources/org/springframework/batch/io/sql/init-foo-schema-hsqldb.sql +++ b/infrastructure/src/test/resources/org/springframework/batch/io/sql/init-foo-schema-hsqldb.sql @@ -11,3 +11,11 @@ INSERT INTO t_foos (id, name, value) VALUES (2, 'bar2', 2); INSERT INTO t_foos (id, name, value) VALUES (3, 'bar3', 3); INSERT INTO t_foos (id, name, value) VALUES (4, 'bar4', 4); INSERT INTO t_foos (id, name, value) VALUES (5, 'bar5', 5); + +CREATE TABLE T_WRITE_FOOS ( + ID BIGINT NOT NULL, + NAME VARCHAR(45), + VALUE INTEGER +); + +ALTER TABLE T_WRITE_FOOS ADD PRIMARY KEY (ID);