From ce399d0555b54416f7302a409fae45ddfa339cf5 Mon Sep 17 00:00:00 2001
From: lucasward
* Each call to read() will call the provided RowMapper, (NOTE: Calling read() * without setting a RowMapper will result in an IllegalStateException!) passing @@ -72,19 +72,19 @@ import org.springframework.util.Assert; * 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 @@ -95,7 +95,7 @@ import org.springframework.util.Assert; * 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 @@ -105,13 +105,13 @@ import org.springframework.util.Assert; * 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 */ @@ -163,7 +163,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, /** * Assert that mandatory properties are set. - * + * * @throws IllegalArgumentException * if either data source or sql properties not set. */ @@ -174,7 +174,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, /** * Public setter for the data source for injection purposes. - * + * * @param dataSource */ public void setDataSource(DataSource dataSource) { @@ -187,7 +187,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, * 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 @@ -216,9 +216,12 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, currentProcessedRow++; } } + + Object mappedResult = mapper.mapRow(rs, currentProcessedRow); + verifyCursorPosition(currentProcessedRow); - return mapper.mapRow(rs, currentProcessedRow); + return mappedResult; } } catch (SQLException se) { throw getExceptionTranslator().translate( @@ -242,7 +245,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, /** * Set the ResultSet's current row to the last marked position. - * + * * @throws DataAccessException */ private void reset() { @@ -265,7 +268,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, * 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() { @@ -280,7 +283,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, /** * 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 { @@ -323,7 +326,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, } catch (SQLException se) { close(); throw getExceptionTranslator().translate("Executing query", - stmt.toString(), se); + getSql(), se); } BatchTransactionSynchronizationManager @@ -335,7 +338,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, * 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 @@ -373,10 +376,10 @@ public class SqlCursorInputSource implements InputSource, DisposableBean, /* * 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 {
@@ -389,14 +392,14 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
+ warningToLog.getMessage() + "]");
warningToLog = warningToLog.getNextWarning();
}
- } else {
+ } else if(warnings != null){
throw new SQLWarningException("Warning not ignored", warnings);
}
}
/*
* (non-Javadoc)
- *
+ *
* @see org.springframework.batch.restart.Restartable#getRestartData()
*/
public RestartData getRestartData() {
@@ -405,7 +408,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/*
* (non-Javadoc)
- *
+ *
* @see org.springframework.batch.restart.Restartable#restoreFrom(org.springframework.batch.restart.RestartData)
*/
public void restoreFrom(RestartData data) {
@@ -415,7 +418,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
if (rs == null) {
executeQuery();
}
-
+
Properties restartProperties = data.getProperties();
if(restartProperties.containsKey(CURRENT_PROCESSED_ROW) == false){
return;
@@ -434,7 +437,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/*
* (non-Javadoc)
- *
+ *
* @see org.springframework.batch.statistics.StatisticsProvider#getStatistics()
*/
public Properties getStatistics() {
@@ -462,7 +465,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
* 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)
@@ -474,7 +477,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/**
* 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)
@@ -488,7 +491,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
* 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
@@ -501,7 +504,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/**
* Set whether SQLWarnings should be ignored (only logged) or exception
* should be thrown.
- *
+ *
* @param ignoreWarnings
* if TRUE, warnings are ignored
*/
@@ -512,7 +515,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/**
* 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
*/
@@ -522,7 +525,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/**
* Set the RowMapper to be used for all calls to read().
- *
+ *
* @param mapper
*/
public void setMapper(RowMapper mapper) {
@@ -533,7 +536,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
* 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) {
@@ -554,7 +557,7 @@ public class SqlCursorInputSource implements InputSource, DisposableBean,
/*
* @param status transaction status
- *
+ *
* @see org.springframework.transaction.support.TransactionSynchronization#afterCompletion(int)
*/
public void afterCompletion(int status) {
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 52f419592..1de768539 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,5 +1,6 @@
package org.springframework.batch.io.sql;
+import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;
@@ -9,100 +10,142 @@ import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationMan
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;
/**
- *
+ *
* @author Lucas Ward
*
*/
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.close();
+ sqlCursorInputSource.destroy();
super.onTearDown();
}
-
+
protected SqlCursorInputSource getNewInputSource(){
-
+
SqlCursorInputSource result = new SqlCursorInputSource();
result.setDataSource(super.getJdbcTemplate().getDataSource());
result.setSql("SELECT * from T_FOOS");
+ result.setIgnoreWarnings(true);
+ result.setVerifyCursorPosition(true);
+
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{
@@ -113,137 +156,181 @@ public class SqlCursorInputSourceIntegrationTests extends AbstractTransactionalD
//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();
+ }
+ }
}