diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxy.java new file mode 100644 index 000000000..49c77c532 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxy.java @@ -0,0 +1,271 @@ +/* + * Copyright 2002-2008 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.item.database; + +import java.io.PrintWriter; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.sql.Connection; +import java.sql.SQLException; + +import javax.sql.DataSource; + +import org.springframework.jdbc.datasource.ConnectionProxy; +import org.springframework.jdbc.datasource.DataSourceUtils; +import org.springframework.jdbc.datasource.SmartDataSource; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +/** + * Implementation of {@link SmartDataSource} that is capable of keeping a single JDBC Connection + * which is NOT closed after each use even if {@link Connection#close()} is called. + * + * The connection can be kept open over multiple transactions when used together with any of Spring's + * {@link org.springframework.transaction.PlatformTransactionManager} implementations. + * + *

Loosely based on the SingleConnectionDataSource implementation in Spring Core. Intended + * to be used with the {@link JdbcCursorItemReader} to provide a connection that remains + * open across transaction boundaries, It remains open for the life of the cursor, and can be + * shared with the main transaction of the rest of the step processing. + * + *

Once close suppression has been turned on for a connection, it will be returned for the first + * {@link #getConnection()} call. Any subsequent calls to {@link #getConnection()} will retrieve a + * new connection from the wrapped {@link DataSource} until the {@link DataSourceUtils} queries + * whether the connection should be closed or not by calling {@link #shouldClose(Connection)} for + * the close-suppressed {@link Connection}. At that point the cycle starts over again, and the next + * {@link #getConnection()} call will have the {@link Connection} that is being close-suppressed + * returned. This allows the use of the close-suppressed {@link Connection} to be the main + * {@link Connection} for an extended data access process. The close suppression is turned off by + * calling {@link #stopCloseSuppression(Connection)}. + * + *

This class is not multi-threading capable. + * + *

The connection returned will be a close-suppressing proxy instead of the physical + * {@link Connection}. Be aware that you will not be able to cast this to a native + * OracleConnection or the like anymore; you need to use a + * {@link org.springframework.jdbc.support.nativejdbc.NativeJdbcExtractor}. + * + * @author Thomas Risberg + * @see #getConnection() + * @see java.sql.Connection#close() + * @see DataSourceUtils#releaseConnection + * @see org.springframework.jdbc.support.nativejdbc.NativeJdbcExtractor + */ +public class ExtendedConnectionDataSourceProxy implements SmartDataSource { + + /** Provided DataSource */ + private DataSource dataSource; + + /** The connection to suppress close calls for */ + private Connection closeSuppressedConnection = null; + + /** The connection to suppress close calls for */ + private boolean borrowedConnection = false; + + /** Synchronization monitor for the shared Connection */ + private final Object connectionMonitor = new Object(); + + /** + * No arg constructor for use when configured using JavaBean style. + */ + public ExtendedConnectionDataSourceProxy() { + } + + /** + * Constructor that takes as a parameter with the {&link DataSource} to be wrapped. + */ + public ExtendedConnectionDataSourceProxy(DataSource dataSource) { + this.dataSource = dataSource; + } + + /** + * Setter for the {&link DataSource} that is to be wrapped. + * + * @param dataSource the DataSource + */ + public void setDataSource(DataSource dataSource) { + this.dataSource = dataSource; + } + + /** + * @see SmartDataSource + */ + public boolean shouldClose(Connection connection) { + boolean shouldClose = !isCloseSuppressionActive(connection); + if (borrowedConnection && closeSuppressedConnection.equals(connection)) { + borrowedConnection = false; + } + return shouldClose; + } + + /** + * Return the status of close suppression being activated for a given {@link Connection} + * + * @param connection the {@link Connection} that the close suppression status is requested for + * @return true or false + */ + public boolean isCloseSuppressionActive(Connection connection) { + if (connection == null) { + return false; + } + return connection.equals(closeSuppressedConnection) ? true : false; + } + + /** + * + * @param connection the {@link Connection} that close suppression is requested for + */ + public void startCloseSuppression(Connection connection) { + synchronized (this.connectionMonitor) { + closeSuppressedConnection = connection; + if(TransactionSynchronizationManager.isActualTransactionActive()) { + borrowedConnection = true; + } + } + } + + /** + * + * @param connection the {@link Connection} that close suppression should be turned off for + */ + public void stopCloseSuppression(Connection connection) { + synchronized (this.connectionMonitor) { + closeSuppressedConnection = null; + borrowedConnection = false; + } + } + + public Connection getConnection() throws SQLException { + synchronized (this.connectionMonitor) { + return initConnection(null, null); + } + } + + public Connection getConnection(String username, String password) throws SQLException { + synchronized (this.connectionMonitor) { + return initConnection(username, password); + } + } + + private boolean completeCloseCall(Connection connection) { + if (borrowedConnection && closeSuppressedConnection.equals(connection)) { + borrowedConnection = false; + } + return isCloseSuppressionActive(connection); + } + + private Connection initConnection(String username, String password) throws SQLException { + if (closeSuppressedConnection != null) { + if (!borrowedConnection) { + borrowedConnection = true; + return closeSuppressedConnection; + } + } + Connection target; + if (username != null) { + target = dataSource.getConnection(username, password); + } + else { + target = dataSource.getConnection(); + } + Connection connection = getCloseSuppressingConnectionProxy(target); + return connection; + } + + public PrintWriter getLogWriter() throws SQLException { + return dataSource.getLogWriter(); + } + + public int getLoginTimeout() throws SQLException { + return dataSource.getLoginTimeout(); + } + + public void setLogWriter(PrintWriter out) throws SQLException { + dataSource.setLogWriter(out); + } + + public void setLoginTimeout(int seconds) throws SQLException { + dataSource.setLoginTimeout(seconds); + } + + /** + * Wrap the given Connection with a proxy that delegates every method call to it + * but suppresses close calls. + * @param target the original Connection to wrap + * @return the wrapped Connection + */ + protected Connection getCloseSuppressingConnectionProxy(Connection target) { + return (Connection) Proxy.newProxyInstance( + ConnectionProxy.class.getClassLoader(), + new Class[] {ConnectionProxy.class}, + new CloseSuppressingInvocationHandler(target, this)); + } + + + /** + * Invocation handler that suppresses close calls on JDBC Connections until the associated instance of the + * ExtendedConnectionDataSourceProxy determines the connection should actually be closed. + */ + private static class CloseSuppressingInvocationHandler implements InvocationHandler { + + private final Connection target; + + private final ExtendedConnectionDataSourceProxy dataSource; + + public CloseSuppressingInvocationHandler(Connection target, ExtendedConnectionDataSourceProxy dataSource) { + this.dataSource = dataSource; + this.target = target; + } + + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + // Invocation on ConnectionProxy interface coming in... + + if (method.getName().equals("equals")) { + // Only consider equal when proxies are identical. + return (proxy == args[0] ? Boolean.TRUE : Boolean.FALSE); + } + else if (method.getName().equals("hashCode")) { + // Use hashCode of Connection proxy. + return new Integer(System.identityHashCode(proxy)); + } + else if (method.getName().equals("close")) { + // Handle close method: don't pass the call on if we are suppressing close calls. + if (dataSource.completeCloseCall((Connection)proxy)) { + return null; + } + else { + target.close(); + return null; + } + } + else if (method.getName().equals("getTargetConnection")) { + // Handle getTargetConnection method: return underlying Connection. + return this.target; + } + + // Invoke method on target Connection. + try { + return method.invoke(this.target, args); + } + catch (InvocationTargetException ex) { + throw ex.getTargetException(); + } + } + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java index 7d9e2de3d..dcf968974 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java @@ -31,6 +31,7 @@ import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.dao.InvalidDataAccessResourceUsageException; import org.springframework.jdbc.SQLWarningException; import org.springframework.jdbc.core.PreparedStatementSetter; @@ -40,15 +41,38 @@ 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.TransactionSynchronizationManager; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; /** *

* Simple item reader that opens a JDBC 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. + * next row in the ResultSet. + *

+ * + *

+ * By default the cursor will be opened using a separate connection. The ResultSet for the cursor + * is held open regardless of commits or roll backs in a surrounding transaction. Clients of this + * reader are responsible for buffering the items in the case that they need to be re-presented on a + * rollback. This buffering is handled by the step implementations provided and is only a concern for + * anyone writing their own step implementations. + *

+ * + *

+ * The statement used to open the cursor is created with the 'READ_ONLY' option since a non read-only + * cursor may unnecessarily lock tables or rows. It is also opened with 'TYPE_FORWARD_ONLY' option. + * By default the cursor will be opened using a separate connection which means that it will not participate + * in any transactions created as part of the step processing. + *

+ * + *

+ * There is an option ({@link #setUseSharedExtendedConnection(boolean)} that will share the connection + * used for the cursor with the rest of the step processing. If you set this flag to true + * then you must wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the + * connection from being closed and released after each commit performed as part of the step processing. + * You must also use a JDBC driver supporting JDBC 3.0 or later since the cursor will be opened with the + * additional option of 'HOLD_CUSORS_OVER_COMMIT' enabled. *

* *

@@ -56,12 +80,10 @@ import org.springframework.util.ClassUtils; * ResultSet. There is currently no wrapping of the ResultSet to suppress calls * to next(). However, if the RowMapper (mistakenly) 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 - * the correct position before returning. The reason for such strictness on the + * position and throw a DataAccessException if it is not. The reason for such strictness on the * ResultSet is due to the need to maintain control for transactions and * restartability. This ensures that each call to {@link #read()} returns the - * ResultSet at the correct line, regardless of rollbacks or restarts. + * ResultSet at the correct row, regardless of rollbacks or restarts. *

* *

@@ -73,30 +95,22 @@ import org.springframework.util.ClassUtils; *

* *

- * Transactions: The same ResultSet is held open regardless of commits or roll - * backs in a surrounding transaction. When a transaction is committed, the - * reader will be notified through the {@link #update(ExecutionContext)} so that - * it can save it's current row number. Clients of this reader are responsible - * for buffering the items in the case that they need to be re-presented on a - * rollback. - *

- * - *

* Calling close on this {@link ItemStream} will cause all resources it is * currently using to be freed. (Connection, ResultSet, etc). It is then illegal - * to call {@link #read()} again until it has been opened. + * to call {@link #read()} again until it has been re-opened. *

* *

* Known limitation: when used with Derby * {@link #setVerifyCursorPosition(boolean)} needs to be false * because {@link ResultSet#getRow()} call used for cursor position verification - * throws an exception. + * is not available for 'TYPE_FORWARD_ONLY' result sets. *

* * @author Lucas Ward * @author Peter Zozom * @author Robert Kasanicky + * @author Thomas Risberg */ public class JdbcCursorItemReader extends AbstractItemCountingItemStreamItemReader implements InitializingBean { @@ -134,7 +148,7 @@ public class JdbcCursorItemReader extends AbstractItemCountingItemStreamItemR private boolean driverSupportsAbsolute = false; - private boolean participateInExistingTransaction = false; + private boolean useSharedExtendedConnection = false; public JdbcCursorItemReader() { setName(ClassUtils.getShortName(JdbcCursorItemReader.class)); @@ -162,25 +176,28 @@ public class JdbcCursorItemReader extends AbstractItemCountingItemStreamItemR } /** - * 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. + * Executes the provided SQL query. */ private void executeQuery() { Assert.state(dataSource != null, "DataSource must not be null."); try { - if (participateInExistingTransaction) { + if (useSharedExtendedConnection) { + if (!(dataSource instanceof ExtendedConnectionDataSourceProxy)) { + throw new InvalidDataAccessApiUsageException( + "You must use a ExtendedConnectionDataSourceProxy for the dataSource when " + + "useSharedExtendedConnection is set to true."); + } this.con = DataSourceUtils.getConnection(dataSource); + ((ExtendedConnectionDataSourceProxy)dataSource).startCloseSuppression(this.con); + preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, + ResultSet.HOLD_CURSORS_OVER_COMMIT); } else { this.con = dataSource.getConnection(); + preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); } - preparedStatement = this.con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, - ResultSet.HOLD_CURSORS_OVER_COMMIT); applyStatementSettings(preparedStatement); if (this.preparedStatementSetter != null) { preparedStatementSetter.setValues(preparedStatement); @@ -385,16 +402,23 @@ public class JdbcCursorItemReader extends AbstractItemCountingItemStreamItemR } /** - * Indicate whether the cursor should be opened as part of an existing transaction or if it - * should be opened in its own transaction. The default is for the cursor to be opened in its - * own transaction. If you set this flag to true then you should wrap the DataSource in a - * {@link org.springframework.jdbc.datasource.SingleConnectionDataSource} to prevent the - * connection from being closed after each commit. + * Indicate whether the connection used for the cursor should be used by all other processing + * thus sharing the same transaction. If this is set to false, which is the default, then the + * cursor will be opened using in its connection and will not participate in any transactions + * started for the rest of the step processing. If you set this flag to true then you must + * wrap the DataSource in a {@link ExtendedConnectionDataSourceProxy} to prevent the + * connection from being closed and released after each commit. + * + * When you set this option to true then the statement used to open the cursor + * will be created with both 'READ_ONLY' and 'HOLD_CUSORS_OVER_COMMIT' options. This allows + * holding the cursor open over transaction start and commits performed in the step processing. + * To use this feature you need a database that supports this and a JDBC driver supporting + * JDBC 3.0 or later. * - * @param participateInExistingTransaction false by default + * @param useSharedExtendedConnection false by default */ - public void setParticipateInExistingTransaction(boolean participateInExistingTransaction) { - this.participateInExistingTransaction = participateInExistingTransaction; + public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) { + this.useSharedExtendedConnection = useSharedExtendedConnection; } /** @@ -415,10 +439,17 @@ public class JdbcCursorItemReader extends AbstractItemCountingItemStreamItemR protected void doClose() throws Exception { initialized = false; JdbcUtils.closeResultSet(this.rs); - JdbcUtils.closeStatement(this.preparedStatement); - JdbcUtils.closeConnection(this.con); rs = null; - + JdbcUtils.closeStatement(this.preparedStatement); + if (useSharedExtendedConnection && dataSource instanceof ExtendedConnectionDataSourceProxy) { + ((ExtendedConnectionDataSourceProxy)dataSource).stopCloseSuppression(this.con); + if (!TransactionSynchronizationManager.isActualTransactionActive()) { + DataSourceUtils.releaseConnection(con, dataSource); + } + } + else { + JdbcUtils.closeConnection(this.con); + } } /** diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxyTests.java new file mode 100644 index 000000000..2892cdb4b --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxyTests.java @@ -0,0 +1,229 @@ +package org.springframework.batch.item.database; + +import static org.easymock.EasyMock.*; +import static org.junit.Assert.*; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; + +import javax.sql.DataSource; + +import org.junit.Test; +import org.junit.internal.runners.JUnit4ClassRunner; +import org.junit.runner.RunWith; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.DataSourceUtils; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +@RunWith(JUnit4ClassRunner.class) +public class ExtendedConnectionDataSourceProxyTests { + + @Test + public void testOperationWithDataSourceUtils() throws SQLException { + Connection con = createMock(Connection.class); + DataSource ds = createMock(DataSource.class); + + expect(ds.getConnection()).andReturn(con); // con1 + con.close(); + expect(ds.getConnection()).andReturn(con); // con2 + con.close(); + + expect(ds.getConnection()).andReturn(con); // con3 + con.close(); // con3 + expect(ds.getConnection()).andReturn(con); // con4 + con.close(); // con4 + + replay(ds); + replay(con); + + final ExtendedConnectionDataSourceProxy csds = new ExtendedConnectionDataSourceProxy(ds); + + Connection con1 = csds.getConnection(); + Connection con2 = csds.getConnection(); + assertNotSame("shouldn't be the same connection", con1, con2); + + assertTrue("should be able to close connection", csds.shouldClose(con1)); + con1.close(); + assertTrue("should be able to close connection", csds.shouldClose(con2)); + con2.close(); + + Connection con3 = csds.getConnection(); + csds.startCloseSuppression(con3); + Connection con3_1 = csds.getConnection(); + assertSame("should be same connection", con3_1, con3); + assertFalse("should not be able to close connection", csds.shouldClose(con3)); + con3_1.close(); // no mock call for this - should be suppressed + Connection con3_2 = csds.getConnection(); + assertSame("should be same connection", con3_2, con3); + Connection con4 = csds.getConnection(); + assertNotSame("shouldn't be same connection", con4, con3); + csds.stopCloseSuppression(con3); + assertTrue("should be able to close connection", csds.shouldClose(con3)); + con3_1 = null; + con3_2 = null; + con3.close(); + assertTrue("should be able to close connection", csds.shouldClose(con4)); + con4.close(); + + verify(ds); + verify(con); + + } + + @Test + public void testOperationWithDirectCloseCall() throws SQLException { + Connection con = createMock(Connection.class); + DataSource ds = createMock(DataSource.class); + + expect(ds.getConnection()).andReturn(con); // con1 + con.close(); + expect(ds.getConnection()).andReturn(con); // con2 + con.close(); + + replay(ds); + replay(con); + + final ExtendedConnectionDataSourceProxy csds = new ExtendedConnectionDataSourceProxy(ds); + + Connection con1 = csds.getConnection(); + csds.startCloseSuppression(con1); + Connection con1_1 = csds.getConnection(); + assertSame("should be same connection", con1_1, con1); + con1_1.close(); // no mock call for this - should be suppressed + Connection con1_2 = csds.getConnection(); + assertSame("should be same connection", con1_2, con1); + Connection con2 = csds.getConnection(); + assertNotSame("shouldn't be same connection", con2, con1); + csds.stopCloseSuppression(con1); + assertTrue("should be able to close connection", csds.shouldClose(con1)); + con1_1 = null; + con1_2 = null; + con1.close(); + assertTrue("should be able to close connection", csds.shouldClose(con2)); + con2.close(); + + verify(ds); + verify(con); + + } + + @Test + public void testSupressOfCloseWithJdbcTemplate() throws Exception { + + Connection con = createMock(Connection.class); + DataSource ds = createMock(DataSource.class); + Statement stmt = createMock(Statement.class); + ResultSet rs = createMock(ResultSet.class); + + // open and start suppressing close + expect(ds.getConnection()).andReturn(con); + + // transaction 1 + expect(con.getAutoCommit()).andReturn(false); + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select baz from bar")).andReturn(rs); + expect(rs.next()).andReturn(false); + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select foo from bar")).andReturn(rs); + expect(rs.next()).andReturn(false); + con.commit(); + + // transaction 2 + expect(con.getAutoCommit()).andReturn(false); + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select ham from foo")).andReturn(rs); + expect(rs.next()).andReturn(false); + // REQUIRES_NEW transaction + expect(ds.getConnection()).andReturn(con); + expect(con.getAutoCommit()).andReturn(false); + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select 1 from eggs")).andReturn(rs); + expect(rs.next()).andReturn(false); + con.commit(); + con.close(); + // resume transaction 2 + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select more, ham from foo")).andReturn(rs); + expect(rs.next()).andReturn(false); + con.commit(); + + // transaction 3 + expect(con.getAutoCommit()).andReturn(false); + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select spam from ham")).andReturn(rs); + expect(rs.next()).andReturn(false); + con.commit(); + + // stop suppressing close and close + con.close(); + + // standalone query + expect(ds.getConnection()).andReturn(con); + expect(con.createStatement()).andReturn(stmt); + expect(stmt.executeQuery("select egg from bar")).andReturn(rs); + expect(rs.next()).andReturn(false); + con.close(); + + replay(rs); + replay(stmt); + replay(con); + replay(ds); + + final ExtendedConnectionDataSourceProxy csds = new ExtendedConnectionDataSourceProxy(); + csds.setDataSource(ds); + PlatformTransactionManager tm = new DataSourceTransactionManager(csds); + TransactionTemplate tt = new TransactionTemplate(tm); + final TransactionTemplate tt2 = new TransactionTemplate(tm); + tt2.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + final JdbcTemplate template = new JdbcTemplate(csds); + + Connection connection = DataSourceUtils.getConnection(csds); + csds.startCloseSuppression(connection); + tt.execute( + new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + template.queryForList("select baz from bar"); + template.queryForList("select foo from bar"); + return null; + } + }); + tt.execute( + new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + template.queryForList("select ham from foo"); + tt2.execute( + new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + template.queryForList("select 1 from eggs"); + return null; + } + }); + template.queryForList("select more, ham from foo"); + return null; + } + }); + tt.execute( + new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + template.queryForList("select spam from ham"); + return null; + } + }); + csds.stopCloseSuppression(connection); + DataSourceUtils.releaseConnection(connection, csds); + template.queryForList("select egg from bar"); + + verify(rs); + verify(stmt); + verify(con); + verify(ds); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java index e819bc2d3..df400f8c8 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java @@ -26,7 +26,33 @@ public class JdbcCursorItemReaderConfigTests { */ @Test public void testUsesCurrentTransaction() throws Exception { - + DataSource ds = createMock(DataSource.class); + Connection con = createMock(Connection.class); + expect(con.getAutoCommit()).andReturn(false); + PreparedStatement ps = createMock(PreparedStatement.class); + expect(con.prepareStatement("select foo from bar", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, + ResultSet.HOLD_CURSORS_OVER_COMMIT)).andReturn(ps); + expect(ds.getConnection()).andReturn(con); + expect(ds.getConnection()).andReturn(con); + con.commit(); + replay(con); + replay(ds); + PlatformTransactionManager tm = new DataSourceTransactionManager(ds); + TransactionTemplate tt = new TransactionTemplate(tm); + final JdbcCursorItemReader reader = new JdbcCursorItemReader(); + reader.setDataSource(new ExtendedConnectionDataSourceProxy(ds)); + reader.setUseSharedExtendedConnection(true); + reader.setSql("select foo from bar"); + final ExecutionContext ec = new ExecutionContext(); + tt.execute( + new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + reader.open(ec); + reader.close(); + return null; + } + }); + verify(ds); } /* @@ -39,8 +65,7 @@ public class JdbcCursorItemReaderConfigTests { Connection con = createMock(Connection.class); expect(con.getAutoCommit()).andReturn(false); PreparedStatement ps = createMock(PreparedStatement.class); - expect(con.prepareStatement("select foo from bar", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, - ResultSet.HOLD_CURSORS_OVER_COMMIT)).andReturn(ps); + expect(con.prepareStatement("select foo from bar", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)).andReturn(ps); expect(ds.getConnection()).andReturn(con); expect(ds.getConnection()).andReturn(con); con.commit();