HHH-5765 : Integrate LogicalConnection into ConnectionManager

hibernate-core/src/main/java/org/hibernate/engine/SessionFactoryImplementor.java

@@ -36,6 +36,7 @@ import org.hibernate.Interceptor;
 import org.hibernate.MappingException;
 import org.hibernate.SessionFactory;
 import org.hibernate.ConnectionReleaseMode;
+import org.hibernate.engine.jdbc.spi.JdbcServices;
 import org.hibernate.engine.jdbc.spi.SQLExceptionHelper;
 import org.hibernate.proxy.EntityNotFoundDelegate;
 import org.hibernate.engine.query.QueryPlanCache;
@@ -97,10 +98,16 @@ public interface SessionFactoryImplementor extends Mapping, SessionFactory {
 	 */
 	public CollectionPersister getCollectionPersister(String role) throws MappingException;
 
+	/**
+	 * Get the JdbcServices.
+	 * @return the JdbcServices
+	 */
+	public JdbcServices getJdbcServices();
+
 	/**
 	 * Get the SQL dialect.
 	 * <p/>
-	 * Shorthand for {@link #getSettings()}.{@link Settings#getDialect()}
+	 * Shorthand for {@link #getJdbcServices().getDialect()}.{@link JdbcServices#getDialect()}
 	 *
 	 * @return The dialect
 	 */

hibernate-core/src/main/java/org/hibernate/engine/jdbc/internal/LogicalConnectionImpl.java

@@ -23,6 +23,9 @@
  */
 package org.hibernate.engine.jdbc.internal;
 
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.ArrayList;
@@ -38,6 +41,7 @@ import org.hibernate.engine.jdbc.spi.JdbcResourceRegistry;
 import org.hibernate.engine.jdbc.spi.JdbcServices;
 import org.hibernate.engine.jdbc.spi.ConnectionObserver;
 import org.hibernate.engine.jdbc.spi.LogicalConnectionImplementor;
+import org.hibernate.jdbc.BorrowedConnectionProxy;
 
 /**
  * LogicalConnectionImpl implementation
@@ -48,6 +52,8 @@ public class LogicalConnectionImpl implements LogicalConnectionImplementor {
 	private static final Logger log = LoggerFactory.getLogger( LogicalConnectionImpl.class );
 
 	private transient Connection physicalConnection;
+	private transient Connection borrowedConnection;
+
 	private final ConnectionReleaseMode connectionReleaseMode;
 	private final JdbcServices jdbcServices;
 	private final JdbcResourceRegistry jdbcResourceRegistry;
@@ -61,19 +67,47 @@ public class LogicalConnectionImpl implements LogicalConnectionImplementor {
 	        Connection userSuppliedConnection,
 	        ConnectionReleaseMode connectionReleaseMode,
 	        JdbcServices jdbcServices) {
-		if ( connectionReleaseMode == ConnectionReleaseMode.AFTER_STATEMENT &&
-				! jdbcServices.getConnectionProvider().supportsAggressiveRelease() ) {
-			log.debug( "connection provider reports to not support aggressive release; overriding" );
-			connectionReleaseMode = ConnectionReleaseMode.AFTER_TRANSACTION;
-		}
 		this.physicalConnection = userSuppliedConnection;
-		this.connectionReleaseMode = connectionReleaseMode;
+		this.connectionReleaseMode =
+				determineConnectionReleaseMode(
+						userSuppliedConnection != null, connectionReleaseMode, jdbcServices
+				);
 		this.jdbcServices = jdbcServices;
 		this.jdbcResourceRegistry = new JdbcResourceRegistryImpl( jdbcServices.getSqlExceptionHelper() );
 
 		this.isUserSuppliedConnection = ( userSuppliedConnection != null );
 	}
 
+	public LogicalConnectionImpl(
+			ConnectionReleaseMode connectionReleaseMode,
+			JdbcServices jdbcServices,
+			boolean isUserSuppliedConnection,
+			boolean isClosed) {
+		this.connectionReleaseMode = determineConnectionReleaseMode(
+				isUserSuppliedConnection, connectionReleaseMode, jdbcServices
+		);
+		this.jdbcServices = jdbcServices;
+		this.jdbcResourceRegistry = new JdbcResourceRegistryImpl( jdbcServices.getSqlExceptionHelper() );
+		this.isUserSuppliedConnection = isUserSuppliedConnection;
+		this.isClosed = isClosed;
+	}
+
+	private static ConnectionReleaseMode determineConnectionReleaseMode(boolean isUserSuppliedConnection,
+																		ConnectionReleaseMode connectionReleaseMode,
+																		JdbcServices jdbcServices) {
+		if ( isUserSuppliedConnection ) {
+			return ConnectionReleaseMode.ON_CLOSE;
+		}
+		else if ( connectionReleaseMode == ConnectionReleaseMode.AFTER_STATEMENT &&
+				! jdbcServices.getConnectionProvider().supportsAggressiveRelease() ) {
+			log.debug( "connection provider reports to not support aggressive release; overriding" );
+			return ConnectionReleaseMode.AFTER_TRANSACTION;
+		}
+		else {
+			return connectionReleaseMode;
+		}
+	}
+
 	/**
 	 * {@inheritDoc}
 	 */
@@ -130,20 +164,25 @@ public class LogicalConnectionImpl implements LogicalConnectionImplementor {
 	 * {@inheritDoc}
 	 */
 	public Connection close() {
-		log.trace( "closing logical connection" );
 		Connection c = physicalConnection;
-		if ( !isUserSuppliedConnection && physicalConnection != null ) {
-			jdbcResourceRegistry.close();
-			releaseConnection();
+		try {
+			releaseBorrowedConnection();
+			log.trace( "closing logical connection" );
+			if ( !isUserSuppliedConnection && physicalConnection != null ) {
+				jdbcResourceRegistry.close();
+				releaseConnection();
+			}
+			return c;
 		}
-		// not matter what
-		physicalConnection = null;
-		isClosed = true;
-		for ( ConnectionObserver observer : observers ) {
-			observer.logicalConnectionClosed();
-		}
-		log.trace( "logical connection closed" );
-		return c;
+		finally {
+			// no matter what
+			physicalConnection = null;
+			isClosed = true;
+			log.trace( "logical connection closed" );
+			for ( ConnectionObserver observer : observers ) {
+				observer.logicalConnectionClosed();
+			}
+		}			
 	}
 
 	/**
@@ -153,6 +192,40 @@ public class LogicalConnectionImpl implements LogicalConnectionImplementor {
 		return connectionReleaseMode;
 	}
 
+	public boolean isUserSuppliedConnection() {
+		return isUserSuppliedConnection;
+	}
+
+	public boolean hasBorrowedConnection() {
+		return borrowedConnection != null;
+	}
+
+	public Connection borrowConnection() {
+		if ( isClosed ) {
+			throw new HibernateException( "connection has been closed" );
+		}
+		if ( isUserSuppliedConnection() ) {
+			return physicalConnection;
+		}
+		else {
+			if ( borrowedConnection == null ) {
+				borrowedConnection = BorrowedConnectionProxy.generateProxy( this );
+			}
+			return borrowedConnection;
+		}
+	}
+
+	public void releaseBorrowedConnection() {
+		if ( borrowedConnection != null ) {
+			try {
+				BorrowedConnectionProxy.renderUnuseable( borrowedConnection );
+			}
+			finally {
+				borrowedConnection = null;
+			}
+		}
+	}
+
 	public void afterStatementExecution() {
 		log.trace( "starting after statement execution processing [{}]", connectionReleaseMode );
 		if ( connectionReleaseMode == ConnectionReleaseMode.AFTER_STATEMENT ) {
@@ -187,7 +260,8 @@ public class LogicalConnectionImpl implements LogicalConnectionImplementor {
 	public void enableReleases() {
 		log.trace( "(re)enabling releases" );
 		releasesEnabled = true;
-		afterStatementExecution();
+		//FIXME: uncomment after new batch stuff is integrated!!!
+		//afterStatementExecution();
 	}
 
 	/**
@@ -232,19 +306,101 @@ public class LogicalConnectionImpl implements LogicalConnectionImplementor {
 	 */
 	private void releaseConnection() throws JDBCException {
 		log.debug( "releasing JDBC connection" );
+		if ( physicalConnection == null ) {
+			return;
+		}
 		try {
-			if ( !physicalConnection.isClosed() ) {
+			if ( ! physicalConnection.isClosed() ) {
 				getJdbcServices().getSqlExceptionHelper().logAndClearWarnings( physicalConnection );
 			}
-			for ( ConnectionObserver observer : observers ) {
-				observer.physicalConnectionReleased();
+			if ( !isUserSuppliedConnection ) {
+				getJdbcServices().getConnectionProvider().closeConnection( physicalConnection );
 			}
-			getJdbcServices().getConnectionProvider().closeConnection( physicalConnection );
-			physicalConnection = null;
 			log.debug( "released JDBC connection" );
 		}
 		catch (SQLException sqle) {
 			throw getJdbcServices().getSqlExceptionHelper().convert( sqle, "Could not close connection" );
 		}
+		finally {
+			physicalConnection = null;
+		}
+		log.debug( "released JDBC connection" );
+		for ( ConnectionObserver observer : observers ) {
+			observer.physicalConnectionReleased();
+		}
 	}
+
+	/**
+	 * Manually disconnect the underlying JDBC Connection.  The assumption here
+	 * is that the manager will be reconnected at a later point in time.
+	 *
+	 * @return The connection mantained here at time of disconnect.  Null if
+	 * there was no connection cached internally.
+	 */
+	public Connection manualDisconnect() {
+		if ( isClosed ) {
+			throw new IllegalStateException( "cannot manually disconnect because logical connection is already closed" );
+		}
+		Connection c = physicalConnection;
+		releaseConnection();
+		return c;
+	}
+
+	/**
+	 * Manually reconnect the underlying JDBC Connection.  Should be called at
+	 * some point after manualDisconnect().
+	 * <p/>
+	 * This form is used for user-supplied connections.
+	 */
+	public void reconnect(Connection suppliedConnection) {
+		if ( isClosed ) {
+			throw new IllegalStateException( "cannot manually reconnect because logical connection is already closed" );
+		}
+		if ( isUserSuppliedConnection ) {
+			if ( suppliedConnection == null ) {
+				throw new IllegalArgumentException( "cannot reconnect a null user-supplied connection" );
+			}
+			else if ( suppliedConnection == physicalConnection ) {
+				log.warn( "reconnecting the same connection that is already connected; should this connection have been disconnected?" );
+			}
+			else if ( physicalConnection != null ) {
+				throw new IllegalArgumentException(
+						"cannot reconnect to a new user-supplied connection because currently connected; must disconnect before reconnecting."
+				);
+			}
+			physicalConnection = suppliedConnection;
+			log.debug( "reconnected JDBC connection" );
+		}
+		else {
+			if ( suppliedConnection != null ) {
+				throw new IllegalStateException( "unexpected user-supplied connection" );
+			}
+			log.debug( "called reconnect() with null connection (not user-supplied)" );
+		}
+	}
+
+	public boolean isReadyForSerialization() {
+		return isUserSuppliedConnection() ?
+				! isPhysicallyConnected() :
+				! getResourceRegistry().hasRegisteredResources()
+				;
+	}
+
+	public void serialize(ObjectOutputStream oos) throws IOException {
+		oos.writeBoolean( isUserSuppliedConnection );
+		oos.writeBoolean( isClosed );
+	}
+
+	public static LogicalConnectionImpl deserialize(
+			ObjectInputStream ois,
+			JdbcServices jdbcServices,
+			ConnectionReleaseMode connectionReleaseMode	) throws IOException {
+		return new LogicalConnectionImpl(
+				connectionReleaseMode,
+				jdbcServices,
+				ois.readBoolean(),
+				ois.readBoolean()
+		);
+ 	}
+
 }

hibernate-core/src/main/java/org/hibernate/engine/jdbc/spi/LogicalConnection.java

@@ -23,6 +23,7 @@
  */
 package org.hibernate.engine.jdbc.spi;
 
+import java.io.Serializable;
 import java.sql.Connection;
 
 /**
@@ -30,7 +31,7 @@ import java.sql.Connection;
  *
  * @author Steve Ebersole
  */
-public interface LogicalConnection {
+public interface LogicalConnection extends Serializable {
 	/**
 	 * Is this logical connection open?  Another phraseology sometimes used is: "are we
 	 * logically connected"?

hibernate-core/src/main/java/org/hibernate/impl/SessionFactoryImpl.java

@@ -711,11 +711,15 @@ public final class SessionFactoryImpl implements SessionFactory, SessionFactoryI
 		return settings;
 	}
 
+	public JdbcServices getJdbcServices() {
+		return serviceRegistry.getService( JdbcServices.class );
+	}
+
 	public Dialect getDialect() {
 		if ( serviceRegistry == null ) {
 			throw new IllegalStateException( "Cannot determine dialect because serviceRegistry is null." );
 		}
-		return serviceRegistry.getService( JdbcServices.class ).getDialect();
+		return getJdbcServices().getDialect();
 	}
 
 	public Interceptor getInterceptor()
@@ -910,10 +914,6 @@ public final class SessionFactoryImpl implements SessionFactory, SessionFactoryI
 		return getEntityPersister(className).getPropertyType(propertyName);
 	}
 
-	private JdbcServices getJdbcServices() {
-		return serviceRegistry.getService( JdbcServices.class );
-	}
-
 	public ConnectionProvider getConnectionProvider() {
 		return serviceRegistry.getService( JdbcServices.class ).getConnectionProvider();
 	}

hibernate-core/src/main/java/org/hibernate/jdbc/AbstractBatcher.java

@@ -43,6 +43,7 @@ import org.hibernate.ScrollMode;
 import org.hibernate.TransactionException;
 import org.hibernate.dialect.Dialect;
 import org.hibernate.engine.SessionFactoryImplementor;
+import org.hibernate.engine.jdbc.internal.LogicalConnectionImpl;
 import org.hibernate.exception.JDBCExceptionHelper;
 import org.hibernate.jdbc.util.FormatStyle;
 import org.hibernate.util.JDBCExceptionReporter;
@@ -580,41 +581,6 @@ public abstract class AbstractBatcher implements Batcher {
 		}
 	}
 
-	public Connection openConnection() throws HibernateException {
-		log.debug("opening JDBC connection");
-		try {
-			return factory.getConnectionProvider().getConnection();
-		}
-		catch (SQLException sqle) {
-			throw factory.getSQLExceptionHelper().convert(
-			        sqle,
-			        "Cannot open connection"
-				);
-		}
-	}
-
-	public void closeConnection(Connection conn) throws HibernateException {
-		if ( conn == null ) {
-			log.debug( "found null connection on AbstractBatcher#closeConnection" );
-			// EARLY EXIT!!!!
-			return;
-		}
-
-		if ( log.isDebugEnabled() ) {
-			log.debug( "closing JDBC connection" + preparedStatementCountsToString() + resultSetCountsToString() );
-		}
-
-		try {
-			if ( !conn.isClosed() ) {
-				JDBCExceptionReporter.logAndClearWarnings( conn );
-			}
-			factory.getConnectionProvider().closeConnection( conn );
-		}
-		catch ( SQLException sqle ) {
-			throw factory.getSQLExceptionHelper().convert( sqle, "Cannot close connection" );
-		}
-	}
-
 	public void cancelLastQuery() throws HibernateException {
 		try {
 			if (lastQuery!=null) lastQuery.cancel();

hibernate-core/src/main/java/org/hibernate/jdbc/Batcher.java

@@ -161,21 +161,6 @@ public interface Batcher {
 
 	public String openResourceStatsAsString();
 
-	// TODO : remove these last two as batcher is no longer managing connections
-
-	/**
-	 * Obtain a JDBC connection
-	 *
-	 * @deprecated Obtain connections from {@link ConnectionProvider} instead
-	 */
-	public Connection openConnection() throws HibernateException;
-	/**
-	 * Dispose of the JDBC connection
-	 *
-	 * @deprecated Obtain connections from {@link ConnectionProvider} instead
-	 */
-	public void closeConnection(Connection conn) throws HibernateException;
-	
 	/**
 	 * Set the transaction timeout to <tt>seconds</tt> later
 	 * than the current system time.

hibernate-core/src/main/java/org/hibernate/jdbc/BatchingBatcher.java

@@ -29,6 +29,8 @@ import java.sql.SQLException;
 
 import org.hibernate.HibernateException;
 import org.hibernate.Interceptor;
+import org.hibernate.engine.SessionFactoryImplementor;
+import org.hibernate.engine.jdbc.internal.LogicalConnectionImpl;
 
 /**
  * An implementation of the <tt>Batcher</tt> interface that

hibernate-core/src/main/java/org/hibernate/jdbc/BorrowedConnectionProxy.java

@@ -25,6 +25,7 @@
 package org.hibernate.jdbc;
 
 import org.hibernate.HibernateException;
+import org.hibernate.engine.jdbc.internal.LogicalConnectionImpl;
 
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
@@ -46,11 +47,11 @@ public class BorrowedConnectionProxy implements InvocationHandler {
 
 	private static final Class[] PROXY_INTERFACES = new Class[] { Connection.class, ConnectionWrapper.class };
 
-	private final ConnectionManager connectionManager;
+	private final LogicalConnectionImpl logicalConnection;
 	private boolean useable = true;
 
-	public BorrowedConnectionProxy(ConnectionManager connectionManager) {
-		this.connectionManager = connectionManager;
+	public BorrowedConnectionProxy(LogicalConnectionImpl logicalConnection) {
+		this.logicalConnection = logicalConnection;
 	}
 
 	/**
@@ -58,7 +59,7 @@ public class BorrowedConnectionProxy implements InvocationHandler {
 	 */
 	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
 		if ( "close".equals( method.getName() ) ) {
-			connectionManager.releaseBorrowedConnection();
+			logicalConnection.releaseBorrowedConnection();
 			return null;
 		}
 		// should probably no-op commit/rollback here, at least in JTA scenarios
@@ -67,11 +68,11 @@ public class BorrowedConnectionProxy implements InvocationHandler {
 		}
 
 		if ( "getWrappedConnection".equals( method.getName() ) ) {
-			return connectionManager.getConnection();
+			return logicalConnection.getConnection();
 		}
 
 		try {
-			return method.invoke( connectionManager.getConnection(), args );
+			return method.invoke( logicalConnection.getConnection(), args );
 		}
 		catch( InvocationTargetException e ) {
 			throw e.getTargetException();
@@ -82,12 +83,12 @@ public class BorrowedConnectionProxy implements InvocationHandler {
 	 * Generates a Connection proxy wrapping the connection managed by the passed
 	 * connection manager.
 	 *
-	 * @param connectionManager The connection manager to wrap with the
+	 * @param logicalConnection The logical connection to wrap with the
 	 * connection proxy.
 	 * @return The generated proxy.
 	 */
-	public static Connection generateProxy(ConnectionManager connectionManager) {
-		BorrowedConnectionProxy handler = new BorrowedConnectionProxy( connectionManager );
+	public static Connection generateProxy(LogicalConnectionImpl logicalConnection) {
+		BorrowedConnectionProxy handler = new BorrowedConnectionProxy( logicalConnection );
 		return ( Connection ) Proxy.newProxyInstance(
 				getProxyClassLoader(),
 		        PROXY_INTERFACES,

hibernate-core/src/main/java/org/hibernate/jdbc/ConnectionManager.java

@@ -37,8 +37,8 @@ import org.hibernate.ConnectionReleaseMode;
 import org.hibernate.HibernateException;
 import org.hibernate.Interceptor;
 import org.hibernate.engine.SessionFactoryImplementor;
+import org.hibernate.engine.jdbc.internal.LogicalConnectionImpl;
 import org.hibernate.engine.jdbc.spi.ConnectionObserver;
-import org.hibernate.util.JDBCExceptionReporter;
 
 /**
  * Encapsulates JDBC Connection management logic needed by Hibernate.
@@ -60,16 +60,11 @@ public class ConnectionManager implements Serializable {
 	private transient SessionFactoryImplementor factory;
 	private final Callback callback;
 
-	private final ConnectionReleaseMode releaseMode;
-	private transient Connection connection;
-	private transient Connection borrowedConnection;
+	private transient LogicalConnectionImpl connection;
 
-	private final boolean wasConnectionSupplied;
 	private transient Batcher batcher;
 	private transient Interceptor interceptor;
-	private boolean isClosed;
-	private transient boolean isFlushing;
- 
+
 	/**
 	 * Constructs a ConnectionManager.
 	 * <p/>
@@ -92,31 +87,35 @@ public class ConnectionManager implements Serializable {
 		this.interceptor = interceptor;
 		this.batcher = factory.getSettings().getBatcherFactory().createBatcher( this, interceptor );
 
-		this.connection = connection;
-		wasConnectionSupplied = ( connection != null );
-
-		this.releaseMode = wasConnectionSupplied ? ConnectionReleaseMode.ON_CLOSE : releaseMode;
+		setupConnection( connection, releaseMode );
 	}
 
 	/**
 	 * Private constructor used exclusively from custom serialization
 	 */
 	private ConnectionManager(
-	        SessionFactoryImplementor factory,
-	        Callback callback,
-	        ConnectionReleaseMode releaseMode,
-	        Interceptor interceptor,
-	        boolean wasConnectionSupplied,
-	        boolean isClosed) {
+			SessionFactoryImplementor factory,
+			Callback callback,
+			ConnectionReleaseMode releaseMode,
+			Interceptor interceptor
+	) {
 		this.factory = factory;
 		this.callback = callback;
 
 		this.interceptor = interceptor;
 		this.batcher = factory.getSettings().getBatcherFactory().createBatcher( this, interceptor );
+	}
 
-		this.wasConnectionSupplied = wasConnectionSupplied;
-		this.isClosed = isClosed;
-		this.releaseMode = wasConnectionSupplied ? ConnectionReleaseMode.ON_CLOSE : releaseMode;
+	private void setupConnection(Connection suppliedConnection,
+								 ConnectionReleaseMode releaseMode
+	) {
+		connection =
+				new LogicalConnectionImpl(
+						suppliedConnection,
+						releaseMode,
+						factory.getJdbcServices()
+				);
+		connection.addObserver( callback );
 	}
 
 	/**
@@ -137,15 +136,6 @@ public class ConnectionManager implements Serializable {
 		return batcher;
 	}
 
-	/**
-	 * Was the connection being used here supplied by the user?
-	 *
-	 * @return True if the user supplied the JDBC connection; false otherwise
-	 */
-	public boolean isSuppliedConnection() {
-		return wasConnectionSupplied;
-	}
-
 	/**
 	 * Retrieves the connection currently managed by this ConnectionManager.
 	 * <p/>
@@ -159,44 +149,20 @@ public class ConnectionManager implements Serializable {
 	 * available (we are currently manually disconnected).
 	 */
 	public Connection getConnection() throws HibernateException {
-		if ( isClosed ) {
-			throw new HibernateException( "connection manager has been closed" );
-		}
-		if ( connection == null  ) {
-			openConnection();
-		}
-		return connection;
+		return connection.getConnection();
 	}
 
 	public boolean hasBorrowedConnection() {
 		// used from testsuite
-		return borrowedConnection != null;
+		return connection.hasBorrowedConnection();
 	}
 
 	public Connection borrowConnection() {
-		if ( isClosed ) {
-			throw new HibernateException( "connection manager has been closed" );
-		}
-		if ( isSuppliedConnection() ) {
-			return connection;
-		}
-		else {
-			if ( borrowedConnection == null ) {
-				borrowedConnection = BorrowedConnectionProxy.generateProxy( this );
-			}
-			return borrowedConnection;
-		}
+		return connection.borrowConnection();
 	}
 
 	public void releaseBorrowedConnection() {
-		if ( borrowedConnection != null ) {
-			try {
-				BorrowedConnectionProxy.renderUnuseable( borrowedConnection );
-			}
-			finally {
-				borrowedConnection = null;
-			}
-		}
+		connection.releaseBorrowedConnection();
 	}
 
 	/**
@@ -208,9 +174,10 @@ public class ConnectionManager implements Serializable {
 	 * @throws SQLException Can be thrown by the Connection.isAutoCommit() check.
 	 */
 	public boolean isAutoCommit() throws SQLException {
-		return connection == null 
-			|| connection.isClosed()
-			|| connection.getAutoCommit();
+		return connection == null ||
+				! connection.isOpen() ||
+				! connection.isPhysicallyConnected() ||
+				connection.getConnection().getAutoCommit();
 	}
 
 	/**
@@ -227,10 +194,10 @@ public class ConnectionManager implements Serializable {
 	 * @return True if the connections will be released after each statement; false otherwise.
 	 */
 	public boolean isAggressiveRelease() {
-		if ( releaseMode == ConnectionReleaseMode.AFTER_STATEMENT ) {
+		if ( connection.getConnectionReleaseMode() == ConnectionReleaseMode.AFTER_STATEMENT ) {
 			return true;
 		}
-		else if ( releaseMode == ConnectionReleaseMode.AFTER_TRANSACTION ) {
+		else if ( connection.getConnectionReleaseMode() == ConnectionReleaseMode.AFTER_TRANSACTION ) {
 			boolean inAutoCommitState;
 			try {
 				inAutoCommitState = isAutoCommit() && ! callback.isTransactionInProgress();
@@ -256,7 +223,7 @@ public class ConnectionManager implements Serializable {
 	 * @return True if the connections will be released after each statement; false otherwise.
 	 */
 	private boolean isAggressiveReleaseNoTransactionCheck() {
-		if ( releaseMode == ConnectionReleaseMode.AFTER_STATEMENT ) {
+		if ( connection.getConnectionReleaseMode() == ConnectionReleaseMode.AFTER_STATEMENT ) {
 			return true;
 		}
 		else {
@@ -268,7 +235,7 @@ public class ConnectionManager implements Serializable {
 				// assume we are in an auto-commit state
 				inAutoCommitState = true;
 			}
-			return releaseMode == ConnectionReleaseMode.AFTER_TRANSACTION && inAutoCommitState;
+			return connection.getConnectionReleaseMode() == ConnectionReleaseMode.AFTER_TRANSACTION && inAutoCommitState;
 		}
 	}
 
@@ -280,7 +247,17 @@ public class ConnectionManager implements Serializable {
 	 * @return True if logically connected; false otherwise.
 	 */
 	public boolean isCurrentlyConnected() {
-		return wasConnectionSupplied ? connection != null : !isClosed;
+		if ( connection != null ) {
+			if ( connection.isUserSuppliedConnection() ) {
+				return connection.isPhysicallyConnected();
+			}
+			else {
+				return connection.isOpen();
+			}
+		}
+		else {
+			return false;
+		}
 	}
 
 	/**
@@ -290,17 +267,11 @@ public class ConnectionManager implements Serializable {
 	 */
 	public void afterStatement() {
 		if ( isAggressiveRelease() ) {
-			if ( isFlushing ) {
-				log.debug( "skipping aggressive-release due to flush cycle" );
-			}
-			else if ( batcher.hasOpenResources() ) {
+			if ( batcher.hasOpenResources() ) {
 				log.debug( "skipping aggresive-release due to open resources on batcher" );
 			}
-			else if ( borrowedConnection != null ) {
-				log.debug( "skipping aggresive-release due to borrowed connection" );
-			}
 			else {
-				aggressiveRelease();
+				connection.afterStatementExecution();
 			}
 		}
 	}
@@ -311,27 +282,33 @@ public class ConnectionManager implements Serializable {
 	 * indicates.
 	 */
 	public void afterTransaction() {
-		if ( isAfterTransactionRelease() ) {
-			aggressiveRelease();
-		}
-		else if ( isAggressiveReleaseNoTransactionCheck() && batcher.hasOpenResources() ) {
-			log.info( "forcing batcher resource cleanup on transaction completion; forgot to close ScrollableResults/Iterator?" );
-			batcher.closeStatements();
-			aggressiveRelease();
-		}
-		else if ( isOnCloseRelease() ) {
-			// log a message about potential connection leaks
-			log.debug( "transaction completed on session with on_close connection release mode; be sure to close the session to release JDBC resources!" );
+		if ( connection != null ) {
+			if ( isAfterTransactionRelease() ) {
+				connection.afterTransaction();
+			}
+			else if ( isAggressiveReleaseNoTransactionCheck() && batcher.hasOpenResources() ) {
+				log.info( "forcing batcher resource cleanup on transaction completion; forgot to close ScrollableResults/Iterator?" );
+				batcher.closeStatements();
+				connection.afterTransaction();
+			}
+			else if ( isOnCloseRelease() ) {
+				// log a message about potential connection leaks
+				log.debug( "transaction completed on session with on_close connection release mode; be sure to close the session to release JDBC resources!" );
+			}
 		}
 		batcher.unsetTransactionTimeout();
 	}
 
 	private boolean isAfterTransactionRelease() {
-		return releaseMode == ConnectionReleaseMode.AFTER_TRANSACTION;
+		return connection.getConnectionReleaseMode() == ConnectionReleaseMode.AFTER_TRANSACTION;
 	}
 
 	private boolean isOnCloseRelease() {
-		return releaseMode == ConnectionReleaseMode.ON_CLOSE;
+		return connection.getConnectionReleaseMode() == ConnectionReleaseMode.ON_CLOSE;
+	}
+
+	public boolean isLogicallyConnected() {
+		return connection != null && connection.isOpen();
 	}
 
 	/**
@@ -342,12 +319,7 @@ public class ConnectionManager implements Serializable {
 	 * there was no connection cached internally.
 	 */
 	public Connection close() {
-		try {
-			return cleanup();
-		}
-		finally {
-			isClosed = true;
-		}
+		return cleanup();
 	}
 
 	/**
@@ -358,7 +330,11 @@ public class ConnectionManager implements Serializable {
 	 * there was no connection cached internally.
 	 */
 	public Connection manualDisconnect() {
-		return cleanup();
+		if ( ! isLogicallyConnected() ) {
+			throw new IllegalStateException( "cannot manually disconnect because not logically connected." );
+		}
+		batcher.closeStatements();
+		return connection.manualDisconnect();
 	}
 
 	/**
@@ -368,6 +344,7 @@ public class ConnectionManager implements Serializable {
 	 * This form is used for ConnectionProvider-supplied connections.
 	 */
 	public void manualReconnect() {
+		manualReconnect( null );
 	}
 
 	/**
@@ -377,7 +354,10 @@ public class ConnectionManager implements Serializable {
 	 * This form is used for user-supplied connections.
 	 */
 	public void manualReconnect(Connection suppliedConnection) {
-		this.connection = suppliedConnection;
+		if ( ! isLogicallyConnected() ) {
+			throw new IllegalStateException( "cannot manually disconnect because not logically connected." );
+		}
+		connection.reconnect( suppliedConnection );
 	}
 
 	/**
@@ -391,93 +371,22 @@ public class ConnectionManager implements Serializable {
 	 * @throws HibernateException
 	 */
 	private Connection cleanup() throws HibernateException {
-		releaseBorrowedConnection();
-
 		if ( connection == null ) {
 			log.trace( "connection already null in cleanup : no action");
 			return null;
 		}
-
 		try {
 			log.trace( "performing cleanup" );
 
-			batcher.closeStatements();
-			Connection c = null;
-			if ( !wasConnectionSupplied ) {
-				closeConnection();
+			if ( isLogicallyConnected() ) {
+				batcher.closeStatements();
 			}
-			else {
-				c = connection;
-			}
-			connection = null;
+			Connection c = connection.close();
 			return c;
 		}
 		finally {
-			callback.physicalConnectionReleased();
-		}
-	}
-
-	/**
-	 * Performs actions required to perform an aggressive release of the
-	 * JDBC Connection.
-	 */
-	private void aggressiveRelease() {
-		if ( !wasConnectionSupplied ) {
-			log.debug( "aggressively releasing JDBC connection" );
-			if ( connection != null ) {
-				closeConnection();
-			}
-		}
-	}
-
-	/**
-	 * Pysically opens a JDBC Connection.
-	 *
-	 * @throws HibernateException
-	 */
-	private void openConnection() throws HibernateException {
-		if ( connection != null ) {
-			return;
-		}
-
-		log.debug("opening JDBC connection");
-		try {
-			connection = factory.getConnectionProvider().getConnection();
-		}
-		catch (SQLException sqle) {
-			throw factory.getSQLExceptionHelper().convert(
-					sqle,
-					"Cannot open connection"
-				);
-		}
-
-		callback.physicalConnectionObtained( connection ); // register synch; stats.connect()
-	}
-
-	/**
-	 * Physically closes the JDBC Connection.
-	 */
-	private void closeConnection() {
-		if ( log.isDebugEnabled() ) {
-			log.debug(
-					"releasing JDBC connection [" +
-					batcher.openResourceStatsAsString() + "]"
-				);
-		}
-
-		try {
-			if ( !connection.isClosed() ) {
-				JDBCExceptionReporter.logAndClearWarnings( connection );
-			}
-			factory.getConnectionProvider().closeConnection( connection );
 			connection = null;
 		}
-		catch (SQLException sqle) {
-			throw factory.getSQLExceptionHelper().convert(
-					sqle, 
-					"Cannot release connection"
-				);
-		}
 	}
 
 	/**
@@ -487,7 +396,7 @@ public class ConnectionManager implements Serializable {
 	 */
 	public void flushBeginning() {
 		log.trace( "registering flush begin" );
-		isFlushing = true;
+		connection.disableReleases();
 	}
 
 	/**
@@ -496,12 +405,12 @@ public class ConnectionManager implements Serializable {
 	 */
 	public void flushEnding() {
 		log.trace( "registering flush end" );
-		isFlushing = false;
+		connection.enableReleases();
 		afterStatement();
 	}
 
 	public boolean isReadyForSerialization() {
-		return wasConnectionSupplied ? connection == null : !batcher.hasOpenResources();
+		return connection == null ? true : ! batcher.hasOpenResources() && connection.isReadyForSerialization();
 	}
 
 	/**
@@ -531,13 +440,10 @@ public class ConnectionManager implements Serializable {
 		factory = (SessionFactoryImplementor) ois.readObject();
 		interceptor = (Interceptor) ois.readObject();
 		ois.defaultReadObject();
-
-		this.batcher = factory.getSettings().getBatcherFactory().createBatcher( this, interceptor );
 	}
 
 	public void serialize(ObjectOutputStream oos) throws IOException {
-		oos.writeBoolean( wasConnectionSupplied );
-		oos.writeBoolean( isClosed );
+		connection.serialize( oos );
 	}
 
 	public static ConnectionManager deserialize(
@@ -546,14 +452,19 @@ public class ConnectionManager implements Serializable {
 	        Interceptor interceptor,
 	        ConnectionReleaseMode connectionReleaseMode,
 	        Callback callback) throws IOException {
-		return new ConnectionManager(
+		ConnectionManager connectionManager = new ConnectionManager(
 				factory,
 		        callback,
 		        connectionReleaseMode,
-		        interceptor,
-		        ois.readBoolean(),
-		        ois.readBoolean()
+		        interceptor
 		);
+		connectionManager.connection =
+				LogicalConnectionImpl.deserialize(
+						ois,
+						factory.getJdbcServices( ),
+						connectionReleaseMode
+				);
+		connectionManager.connection.addObserver( callback );
+		return connectionManager;
 	}
-
 }

hibernate-core/src/main/java/org/hibernate/jdbc/NonBatchingBatcher.java

@@ -29,6 +29,8 @@ import java.sql.SQLException;
 
 import org.hibernate.HibernateException;
 import org.hibernate.Interceptor;
+import org.hibernate.engine.SessionFactoryImplementor;
+import org.hibernate.engine.jdbc.internal.LogicalConnectionImpl;
 
 /**
  * An implementation of the <tt>Batcher</tt> interface that does no batching

hibernate-core/src/test/java/org/hibernate/test/jdbc/proxies/AggressiveReleaseTest.java

@@ -35,6 +35,7 @@ import org.hibernate.ConnectionReleaseMode;
 import org.hibernate.engine.jdbc.internal.LogicalConnectionImpl;
 import org.hibernate.engine.jdbc.spi.ConnectionObserver;
 import org.hibernate.engine.jdbc.internal.proxy.ProxyBuilder;
+import org.hibernate.stat.ConcurrentStatisticsImpl;
 import org.hibernate.test.common.BasicTestingJdbcServiceImpl;
 import org.hibernate.testing.junit.UnitTestCase;