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
index 49c77c532..835b5044c 100644
--- 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
@@ -26,47 +26,60 @@ import java.sql.SQLException;
import javax.sql.DataSource;
+import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.datasource.ConnectionProxy;
import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.datasource.SmartDataSource;
import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.util.Assert;
/**
- * 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.
+ * 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.
+ * 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)}.
+ *
+ * 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.
*
- *
This class is not multi-threading capable.
+ *
+ * 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}.
*
- *
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 {
+public class ExtendedConnectionDataSourceProxy implements SmartDataSource, InitializingBean {
/** Provided DataSource */
private DataSource dataSource;
@@ -87,7 +100,8 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
}
/**
- * Constructor that takes as a parameter with the {&link DataSource} to be wrapped.
+ * Constructor that takes as a parameter with the {&link DataSource} to be
+ * wrapped.
*/
public ExtendedConnectionDataSourceProxy(DataSource dataSource) {
this.dataSource = dataSource;
@@ -114,9 +128,11 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
}
/**
- * Return the status of close suppression being activated for a given {@link Connection}
+ * 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
+ * @param connection the {@link Connection} that the close suppression
+ * status is requested for
* @return true or false
*/
public boolean isCloseSuppressionActive(Connection connection) {
@@ -125,23 +141,25 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
}
return connection.equals(closeSuppressedConnection) ? true : false;
}
-
+
/**
*
- * @param connection the {@link Connection} that close suppression is requested for
+ * @param connection the {@link Connection} that close suppression is
+ * requested for
*/
public void startCloseSuppression(Connection connection) {
synchronized (this.connectionMonitor) {
closeSuppressedConnection = connection;
- if(TransactionSynchronizationManager.isActualTransactionActive()) {
+ if (TransactionSynchronizationManager.isActualTransactionActive()) {
borrowedConnection = true;
}
}
}
-
+
/**
*
- * @param connection the {@link Connection} that close suppression should be turned off for
+ * @param connection the {@link Connection} that close suppression should be
+ * turned off for
*/
public void stopCloseSuppression(Connection connection) {
synchronized (this.connectionMonitor) {
@@ -149,7 +167,7 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
borrowedConnection = false;
}
}
-
+
public Connection getConnection() throws SQLException {
synchronized (this.connectionMonitor) {
return initConnection(null, null);
@@ -204,29 +222,27 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
}
/**
- * Wrap the given Connection with a proxy that delegates every method call to it
- * but suppresses close calls.
+ * 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));
+ 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.
+ * 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;
@@ -244,8 +260,9 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
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)) {
+ // Handle close method: don't pass the call on if we are
+ // suppressing close calls.
+ if (dataSource.completeCloseCall((Connection) proxy)) {
return null;
}
else {
@@ -254,7 +271,8 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
}
}
else if (method.getName().equals("getTargetConnection")) {
- // Handle getTargetConnection method: return underlying Connection.
+ // Handle getTargetConnection method: return underlying
+ // Connection.
return this.target;
}
@@ -267,5 +285,39 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource {
}
}
}
-
+
+ /**
+ * Performs only a 'shallow' non-recursive check of self's and delegate's
+ * class to retain Java 5 compatibility.
+ */
+ public boolean isWrapperFor(Class iface) throws SQLException {
+ if (iface.isAssignableFrom(SmartDataSource.class) || iface.isAssignableFrom(dataSource.getClass())) {
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Returns either self or delegate (in this order) if one of them can be
+ * cast to supplied parameter class. Does *not* support recursive unwrapping
+ * of the delegate to retain Java 5 compatibility.
+ */
+ public T unwrap(Class iface) throws SQLException {
+ if (iface.isAssignableFrom(SmartDataSource.class)) {
+ @SuppressWarnings("unchecked")
+ T casted = (T) this;
+ return casted;
+ }
+ else if (iface.isAssignableFrom(dataSource.getClass())) {
+ @SuppressWarnings("unchecked")
+ T casted = (T) dataSource;
+ return casted;
+ }
+ throw new SQLException("Unsupported class " + iface.getSimpleName());
+ }
+
+ public void afterPropertiesSet() throws Exception {
+ Assert.notNull(dataSource);
+ }
+
}
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
index 2892cdb4b..67fcf46ea 100644
--- 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
@@ -3,6 +3,7 @@ package org.springframework.batch.item.database;
import static org.easymock.EasyMock.*;
import static org.junit.Assert.*;
+import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -16,6 +17,7 @@ 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.jdbc.datasource.SmartDataSource;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
@@ -31,24 +33,24 @@ public class ExtendedConnectionDataSourceProxyTests {
DataSource ds = createMock(DataSource.class);
expect(ds.getConnection()).andReturn(con); // con1
- con.close();
+ con.close();
expect(ds.getConnection()).andReturn(con); // con2
con.close();
-
- expect(ds.getConnection()).andReturn(con); // con3
+
+ expect(ds.getConnection()).andReturn(con); // con3
con.close(); // con3
- expect(ds.getConnection()).andReturn(con); // con4
+ 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));
@@ -71,27 +73,27 @@ public class ExtendedConnectionDataSourceProxyTests {
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();
+ 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();
@@ -108,7 +110,7 @@ public class ExtendedConnectionDataSourceProxyTests {
con1.close();
assertTrue("should be able to close connection", csds.shouldClose(con2));
con2.close();
-
+
verify(ds);
verify(con);
@@ -116,7 +118,7 @@ public class ExtendedConnectionDataSourceProxyTests {
@Test
public void testSupressOfCloseWithJdbcTemplate() throws Exception {
-
+
Connection con = createMock(Connection.class);
DataSource ds = createMock(DataSource.class);
Statement stmt = createMock(Statement.class);
@@ -175,7 +177,7 @@ public class ExtendedConnectionDataSourceProxyTests {
replay(stmt);
replay(con);
replay(ds);
-
+
final ExtendedConnectionDataSourceProxy csds = new ExtendedConnectionDataSourceProxy();
csds.setDataSource(ds);
PlatformTransactionManager tm = new DataSourceTransactionManager(csds);
@@ -183,39 +185,35 @@ public class ExtendedConnectionDataSourceProxyTests {
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() {
+ 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 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");
+ 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");
@@ -226,4 +224,112 @@ public class ExtendedConnectionDataSourceProxyTests {
verify(ds);
}
+ @Test(expected = IllegalArgumentException.class)
+ public void delegateIsRequired() throws Exception {
+
+ ExtendedConnectionDataSourceProxy tested = new ExtendedConnectionDataSourceProxy(null);
+ tested.afterPropertiesSet();
+ }
+
+ @Test
+ public void unwrapForUnsupportedInterface() throws Exception {
+
+ ExtendedConnectionDataSourceProxy tested = new ExtendedConnectionDataSourceProxy(new DataSourceStub());
+
+ assertFalse(tested.isWrapperFor(Unsupported.class));
+
+ try {
+ tested.unwrap(Unsupported.class);
+ fail();
+ }
+ catch (SQLException expected) {
+// this would be the correct behavior in a Java6-only recursive implementation
+// assertEquals(DataSourceStub.UNWRAP_ERROR_MESSAGE, expected.getMessage());
+ assertEquals("Unsupported class " + Unsupported.class.getSimpleName(), expected.getMessage());
+ }
+ }
+
+ @Test
+ public void unwrapForSupportedInterface() throws Exception {
+
+ DataSourceStub ds = new DataSourceStub();
+ ExtendedConnectionDataSourceProxy tested = new ExtendedConnectionDataSourceProxy(ds);
+
+ assertTrue(tested.isWrapperFor(Supported.class));
+ assertEquals(ds, tested.unwrap(Supported.class));
+ }
+
+ @Test
+ public void unwrapForSmartDataSource() throws Exception {
+
+ ExtendedConnectionDataSourceProxy tested = new ExtendedConnectionDataSourceProxy(new DataSourceStub());
+
+ assertTrue(tested.isWrapperFor(DataSource.class));
+ assertEquals(tested, tested.unwrap(DataSource.class));
+
+ assertTrue(tested.isWrapperFor(SmartDataSource.class));
+ assertEquals(tested, tested.unwrap(SmartDataSource.class));
+ }
+
+ /**
+ * Interface implemented by the wrapped DataSource
+ */
+ private static interface Supported {
+ }
+
+ /**
+ * Interface *not* implemented by the wrapped DataSource
+ */
+ private static interface Unsupported {
+ }
+
+ /**
+ * Stub for a wrapped DataSource that implements additional interface. Its
+ * purpose is testing of {@link DataSource#isWrapperFor(Class)} and
+ * {@link DataSource#unwrap(Class)} methods.
+ */
+ private static class DataSourceStub implements DataSource, Supported {
+
+ private static final String UNWRAP_ERROR_MESSAGE = "supplied type is not implemented by this class";
+
+ public Connection getConnection() throws SQLException {
+ throw new UnsupportedOperationException();
+ }
+
+ public Connection getConnection(String username, String password) throws SQLException {
+ throw new UnsupportedOperationException();
+ }
+
+ public PrintWriter getLogWriter() throws SQLException {
+ throw new UnsupportedOperationException();
+ }
+
+ public int getLoginTimeout() throws SQLException {
+ throw new UnsupportedOperationException();
+ }
+
+ public void setLogWriter(PrintWriter out) throws SQLException {
+ throw new UnsupportedOperationException();
+ }
+
+ public void setLoginTimeout(int seconds) throws SQLException {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean isWrapperFor(Class iface) throws SQLException {
+ if (iface.equals(Supported.class) || (iface.equals(DataSource.class))) {
+ return true;
+ }
+ return false;
+ }
+
+ @SuppressWarnings("unchecked")
+ public T unwrap(Class iface) throws SQLException {
+ if (iface.equals(Supported.class) || iface.equals(DataSource.class)) {
+ return (T) this;
+ }
+ throw new SQLException(UNWRAP_ERROR_MESSAGE);
+ }
+
+ }
}