/*
 * Distributed as part of c3p0 v.0.9.5.3
 *
 * Copyright (C) 2018 Machinery For Change, Inc.
 *
 * Author: Steve Waldman <swaldman@mchange.com>
 *
 * This library is free software; you can redistribute it and/or modify
 * it under the terms of EITHER:
 *
 *     1) The GNU Lesser General Public License (LGPL), version 2.1, as 
 *        published by the Free Software Foundation
 *
 * OR
 *
 *     2) The Eclipse Public License (EPL), version 1.0
 *
 * You may choose which license to accept if you wish to redistribute
 * or modify this work. You may offer derivatives of this work
 * under the license you have chosen, or you may provide the same
 * choice of license which you have been offered here.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received copies of both LGPL v2.1 and EPL v1.0
 * along with this software; see the files LICENSE-EPL and LICENSE-LGPL.
 * If not, the text of these licenses are currently available at
 *
 * LGPL v2.1: http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html
 *  EPL v1.0: http://www.eclipse.org/org/documents/epl-v10.php 
 * 
 */

package com.mchange.v2.c3p0.impl;

import com.mchange.v2.c3p0.stmt.*;
import com.mchange.v2.c3p0.ConnectionCustomizer;
import com.mchange.v2.c3p0.SQLWarnings;
import com.mchange.v2.c3p0.UnifiedConnectionTester;
import com.mchange.v2.c3p0.WrapperConnectionPoolDataSource;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.util.LinkedList;
import java.util.WeakHashMap;

import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;

import com.mchange.v1.db.sql.ConnectionUtils;
import com.mchange.v2.async.AsynchronousRunner;
import com.mchange.v2.async.ThreadPoolAsynchronousRunner;
import com.mchange.v2.log.MLevel;
import com.mchange.v2.log.MLog;
import com.mchange.v2.log.MLogger;
import com.mchange.v2.c3p0.ConnectionTester;
import com.mchange.v2.c3p0.QueryConnectionTester;
import com.mchange.v2.resourcepool.CannotAcquireResourceException;
import com.mchange.v2.resourcepool.ResourcePool;
import com.mchange.v2.resourcepool.ResourcePoolException;
import com.mchange.v2.resourcepool.ResourcePoolFactory;
import com.mchange.v2.resourcepool.TimeoutException;
import com.mchange.v2.sql.SqlUtils;

public final class C3P0PooledConnectionPool
{
    private final static boolean ASYNCHRONOUS_CONNECTION_EVENT_LISTENER = false;

    private final static Throwable[] EMPTY_THROWABLE_HOLDER = new Throwable[1];

    final static MLogger logger = MLog.getLogger( C3P0PooledConnectionPool.class );

    final ResourcePool rp;
    final ConnectionEventListener cl = new ConnectionEventListenerImpl();

    final ConnectionTester     connectionTester;
    final GooGooStatementCache scache;
    
    final boolean c3p0PooledConnections;
    final boolean effectiveStatementCache; //configured for caching and using c3p0 pooled Connections

    final int checkoutTimeout;

    final AsynchronousRunner sharedTaskRunner;
    final AsynchronousRunner deferredStatementDestroyer;;

    final ThrowableHolderPool thp = new ThrowableHolderPool();

    final InUseLockFetcher inUseLockFetcher;

    public int getStatementDestroyerNumConnectionsInUse()                           { return scache == null ? -1 : scache.getStatementDestroyerNumConnectionsInUse(); }
    public int getStatementDestroyerNumConnectionsWithDeferredDestroyStatements()   { return scache == null ? -1 : scache.getStatementDestroyerNumConnectionsWithDeferredDestroyStatements(); }
    public int getStatementDestroyerNumDeferredDestroyStatements()                  { return scache == null ? -1 : scache.getStatementDestroyerNumDeferredDestroyStatements(); } 

    
 This "lock fetcher" crap is a lot of ado about very little.
 In theory, there is a hazard that pool maintenance tasks could
 get sufficiently backed up in the Thread pool that say, multple
 Connection tests might run simultaneously. That would be okay,
 but the Statement cache's "mark-in-use" functionality doesn't
 track nested use of resources. So we enforce exclusive, sequential
 execution of internal tests by requiring the tests hold a lock.
 But what lock to hold? The obvious choice is the tested resource's
 lock, but NewPooledConnection is designed for use by clients that
 do not hold its lock. So, we give NewPooledConnection an internal
 Object, an "inInternalUseLock", and lock on this resource instead.
/**
     *  This "lock fetcher" crap is a lot of ado about very little.
     *  In theory, there is a hazard that pool maintenance tasks could
     *  get sufficiently backed up in the Thread pool that say, multple
     *  Connection tests might run simultaneously. That would be okay,
     *  but the Statement cache's "mark-in-use" functionality doesn't
     *  track nested use of resources. So we enforce exclusive, sequential
     *  execution of internal tests by requiring the tests hold a lock.
     *  But what lock to hold? The obvious choice is the tested resource's
     *  lock, but NewPooledConnection is designed for use by clients that
     *  do not hold its lock. So, we give NewPooledConnection an internal
     *  Object, an "inInternalUseLock", and lock on this resource instead.
     */

    private interface InUseLockFetcher
    {
	public Object getInUseLock(Object resc);
    }

    private static class ResourceItselfInUseLockFetcher implements InUseLockFetcher
    {
	public Object getInUseLock(Object resc) { return resc; }
    }

    private static class C3P0PooledConnectionNestedLockLockFetcher implements InUseLockFetcher
    {
	public Object getInUseLock(Object resc)
	{ return ((AbstractC3P0PooledConnection) resc).inInternalUseLock; }
    }

    private static InUseLockFetcher RESOURCE_ITSELF_IN_USE_LOCK_FETCHER = new ResourceItselfInUseLockFetcher();
    private static InUseLockFetcher C3P0_POOLED_CONNECION_NESTED_LOCK_LOCK_FETCHER = new C3P0PooledConnectionNestedLockLockFetcher();

    C3P0PooledConnectionPool( final ConnectionPoolDataSource cpds,
                    final DbAuth auth,
                    int min, 
                    int max, 
                    int start,
                    int inc,
                    int acq_retry_attempts,
                    int acq_retry_delay,
                    boolean break_after_acq_failure,
                    int checkoutTimeout, //milliseconds
                    int idleConnectionTestPeriod, //seconds
                    int maxIdleTime, //seconds
                    int maxIdleTimeExcessConnections, //seconds
                    int maxConnectionAge, //seconds
                    int propertyCycle, //seconds
                    int unreturnedConnectionTimeout, //seconds
                    boolean debugUnreturnedConnectionStackTraces,
                    boolean forceSynchronousCheckins,
                    final boolean testConnectionOnCheckout,
                    final boolean testConnectionOnCheckin,
                    int maxStatements,
                    int maxStatementsPerConnection,
		    /* boolean statementCacheDeferredClose,      */
                    final ConnectionTester connectionTester,
                    final ConnectionCustomizer connectionCustomizer,
                    final String testQuery,
                    final ResourcePoolFactory fact,
                    ThreadPoolAsynchronousRunner taskRunner,
		    ThreadPoolAsynchronousRunner deferredStatementDestroyer,
                    final String parentDataSourceIdentityToken) throws SQLException
                    {
        try
        {
            if (maxStatements > 0 && maxStatementsPerConnection > 0)
                this.scache = new DoubleMaxStatementCache( taskRunner, deferredStatementDestroyer, maxStatements, maxStatementsPerConnection );
            else if (maxStatementsPerConnection > 0)
                this.scache = new PerConnectionMaxOnlyStatementCache( taskRunner, deferredStatementDestroyer, maxStatementsPerConnection );
            else if (maxStatements > 0)
                this.scache = new GlobalMaxOnlyStatementCache( taskRunner, deferredStatementDestroyer, maxStatements );
            else
                this.scache = null;

            this.connectionTester = connectionTester;

            this.checkoutTimeout = checkoutTimeout;

            this.sharedTaskRunner = taskRunner;
	    this.deferredStatementDestroyer = deferredStatementDestroyer;
            
            this.c3p0PooledConnections = (cpds instanceof WrapperConnectionPoolDataSource);
            this.effectiveStatementCache = c3p0PooledConnections && (scache != null);

	    this.inUseLockFetcher = (c3p0PooledConnections ? C3P0_POOLED_CONNECION_NESTED_LOCK_LOCK_FETCHER : RESOURCE_ITSELF_IN_USE_LOCK_FETCHER);

            class PooledConnectionResourcePoolManager implements ResourcePool.Manager
            {	
                //SynchronizedIntHolder totalOpenedCounter  = new SynchronizedIntHolder();
                //SynchronizedIntHolder connectionCounter   = new SynchronizedIntHolder();
                //SynchronizedIntHolder failedCloseCounter  = new SynchronizedIntHolder();

                final boolean connectionTesterIsDefault = (connectionTester instanceof DefaultConnectionTester);
 
                public Object acquireResource() throws Exception
                { 
                    PooledConnection out;

                    if ( connectionCustomizer == null)
                    {
                        out = (auth.equals( C3P0ImplUtils.NULL_AUTH ) ?
                               cpds.getPooledConnection() :
                               cpds.getPooledConnection( auth.getUser(), 
                                                         auth.getPassword() ) );
                    }
                    else
                    {
                        try
                        { 
                            WrapperConnectionPoolDataSourceBase wcpds = (WrapperConnectionPoolDataSourceBase) cpds;

                            out = (auth.equals( C3P0ImplUtils.NULL_AUTH ) ?
                                   wcpds.getPooledConnection( connectionCustomizer, parentDataSourceIdentityToken ) :
                                   wcpds.getPooledConnection( auth.getUser(), 
                                                              auth.getPassword(),
                                                              connectionCustomizer, parentDataSourceIdentityToken ) );
                        }
                        catch (ClassCastException e)
                        {
                            throw SqlUtils.toSQLException("Cannot use a ConnectionCustomizer with a non-c3p0 ConnectionPoolDataSource." +
                                            " ConnectionPoolDataSource: " + cpds.getClass().getName(), e);
                        }
                    }

                    //connectionCounter.increment(); 
                    //totalOpenedCounter.increment();

                    try
                    {
                        if (scache != null)
                        {
                            if (c3p0PooledConnections)
                                ((AbstractC3P0PooledConnection) out).initStatementCache(scache);
                            else
                            {
                                // System.err.print("Warning! StatementPooling not ");
                                // System.err.print("implemented for external (non-c3p0) ");
                                // System.err.println("ConnectionPoolDataSources.");

                                logger.warning("StatementPooling not " +
                                                "implemented for external (non-c3p0) " +
                                "ConnectionPoolDataSources.");
                            }
                        }
                        
                        // log and clear any SQLWarnings present upon acquisition
                        Connection con = null;
                        try
                        {
                            waitMarkPooledConnectionInUse(out);
                            con = out.getConnection();
                            SQLWarnings.logAndClearWarnings( con );
                        }
                        finally
                        {
                            //invalidate the proxy Connection
                            ConnectionUtils.attemptClose( con );

                            unmarkPooledConnectionInUse(out);
                        }
                        
                        return out;
                    }
                    catch (Exception e)
                    {
                        if (logger.isLoggable( MLevel.WARNING ))
                            logger.log(MLevel.WARNING,
				       "A PooledConnection was acquired, but an Exception occurred while preparing it for use. Attempting to destroy.",
				       e);
                        try { destroyResource( out, false ); }
                        catch (Exception e2)
                        {
                            if (logger.isLoggable( MLevel.WARNING ))
                                logger.log( MLevel.WARNING, 
                                                "An Exception occurred while trying to close partially acquired PooledConnection.",
                                                e2 );
                        }

                        throw e;
                    }
                    finally
                    {
                        if (logger.isLoggable( MLevel.FINEST ))
                            logger.finest(this + ".acquireResource() returning. " );
                        //"Currently open Connections: " + connectionCounter.getValue() +
                        //"; Failed close count: " + failedCloseCounter.getValue() +
                        //"; Total processed by this pool: " + totalOpenedCounter.getValue());
                    }
                }

                // REFURBISHMENT:
                // the PooledConnection refurbishes itself when 
                // its Connection view is closed, prior to being
                // checked back in to the pool. But we still may want to
                // test to make sure it is still good.

                public void refurbishResourceOnCheckout( Object resc ) throws Exception
                {
		    synchronized (inUseLockFetcher.getInUseLock(resc))
		    {
			if ( connectionCustomizer != null )
			{
			    Connection physicalConnection = null;
			    try
			    { 
				physicalConnection =  ((AbstractC3P0PooledConnection) resc).getPhysicalConnection();
				waitMarkPhysicalConnectionInUse( physicalConnection );
				if ( testConnectionOnCheckout )
				{
				    if ( Debug.DEBUG && logger.isLoggable( MLevel.FINER ) )
					finerLoggingTestPooledConnection( resc, "CHECKOUT" );
				    else
					testPooledConnection( resc );
				}
				connectionCustomizer.onCheckOut( physicalConnection, parentDataSourceIdentityToken );
			    }
			    catch (ClassCastException e)
			    {
				throw SqlUtils.toSQLException("Cannot use a ConnectionCustomizer with a non-c3p0 PooledConnection." +
							      " PooledConnection: " + resc + 
							      "; ConnectionPoolDataSource: " + cpds.getClass().getName(), e);
			    }
			    finally
			    { unmarkPhysicalConnectionInUse(physicalConnection); }
			}
			else
			{
			    if ( testConnectionOnCheckout )
			    {
				PooledConnection pc = (PooledConnection) resc;
				try
				{
				    waitMarkPooledConnectionInUse( pc );
				    assert !Boolean.FALSE.equals(pooledConnectionInUse( pc )); //null or true are okay

				    if ( Debug.DEBUG && logger.isLoggable( MLevel.FINER ) )
					finerLoggingTestPooledConnection( pc, "CHECKOUT" );
				    else
					testPooledConnection( pc );
				}
				finally
				{ 
				    unmarkPooledConnectionInUse(pc); 
				}
			    }
			}
		    }
                }

		// TODO: refactor this by putting the connectionCustomizer if logic inside the (currently repeated) logic
                public void refurbishResourceOnCheckin( Object resc ) throws Exception
                {
		    Connection proxyToClose = null; // can't close a proxy while we own parent PooledConnection's lock.
		    try
		    {
		      synchronized (inUseLockFetcher.getInUseLock(resc))
		      {
			if ( connectionCustomizer != null )
			{
			    Connection physicalConnection = null;
			    try
			    { 
				physicalConnection =  ((AbstractC3P0PooledConnection) resc).getPhysicalConnection();
                            
				// so by the time we are checked in, all marked-for-destruction statements should be closed.
				waitMarkPhysicalConnectionInUse( physicalConnection );
				connectionCustomizer.onCheckIn( physicalConnection, parentDataSourceIdentityToken );
				SQLWarnings.logAndClearWarnings( physicalConnection );

				if ( testConnectionOnCheckin )
				{ 
				    if ( Debug.DEBUG && logger.isLoggable( MLevel.FINER ) )
					finerLoggingTestPooledConnection( resc, "CHECKIN" );
				    else
					testPooledConnection( resc );
				}

			    }
			    catch (ClassCastException e)
			    {
				throw SqlUtils.toSQLException("Cannot use a ConnectionCustomizer with a non-c3p0 PooledConnection." +
							      " PooledConnection: " + resc + 
							      "; ConnectionPoolDataSource: " + cpds.getClass().getName(), e);
			    }
			    finally
			    { unmarkPhysicalConnectionInUse(physicalConnection); }
			}
			else
			{
			    PooledConnection pc = (PooledConnection) resc;
			    Connection con = null;

			    try
			    {

				// so by the time we are checked in, all marked-for-destruction statements should be closed.
				waitMarkPooledConnectionInUse( pc );
				con = pc.getConnection();
				SQLWarnings.logAndClearWarnings(con);

				if ( testConnectionOnCheckin )
				{ 
				    if ( Debug.DEBUG && logger.isLoggable( MLevel.FINER ) )
					finerLoggingTestPooledConnection( resc, con, "CHECKIN" );
				    else
					testPooledConnection( resc, con );
				}

			    }
			    finally
			    {
				proxyToClose = con;
				unmarkPooledConnectionInUse( pc );
			    }
			}
		      }
		    }
		    finally
		    {
			// close any opened proxy Connection
			ConnectionUtils.attemptClose(proxyToClose);
		    }
                }

                public void refurbishIdleResource( Object resc ) throws Exception
                { 
		    synchronized (inUseLockFetcher.getInUseLock(resc))
		    {
			PooledConnection pc = (PooledConnection) resc;		    
			try
			{
			    waitMarkPooledConnectionInUse( pc );
			    if ( Debug.DEBUG && logger.isLoggable( MLevel.FINER ) )
				finerLoggingTestPooledConnection( resc, "IDLE CHECK" );
			    else
				testPooledConnection( resc );
			}
			finally
			{ unmarkPooledConnectionInUse( pc ); }
		    }
                }

                private void finerLoggingTestPooledConnection(Object resc, String testImpetus) throws Exception
		{ finerLoggingTestPooledConnection( resc, null, testImpetus); }


                private void finerLoggingTestPooledConnection(Object resc, Connection proxyConn, String testImpetus) throws Exception
                {
                    logger.finer("Testing PooledConnection [" + resc + "] on " + testImpetus + ".");
                    try
                    {
                        testPooledConnection( resc, proxyConn );
                        logger.finer("Test of PooledConnection [" + resc + "] on " + testImpetus + " has SUCCEEDED.");
                    }
                    catch (Exception e)
                    {
                        logger.log(MLevel.FINER, "Test of PooledConnection [" + resc + "] on "+testImpetus+" has FAILED.", e);
                        e.fillInStackTrace();
                        throw e;
                    }
                }

                private void testPooledConnection(Object resc) throws Exception
		{ testPooledConnection( resc, null ); }

                private void testPooledConnection(Object resc, Connection proxyConn) throws Exception
                { 
                    PooledConnection pc = (PooledConnection) resc;
		    assert !Boolean.FALSE.equals(pooledConnectionInUse( pc )); //null or true are okay

                    Throwable[] throwableHolder = EMPTY_THROWABLE_HOLDER;
                    int status;
                    Connection openedConn = null;
                    Throwable rootCause = null;
                    try	
                    { 
			// No! Connection must be maked in use PRIOR TO Connection test
                        //waitMarkPooledConnectionInUse( pc ); 
                        
                        // if this is a c3p0 pooled-connection, let's get underneath the
                        // proxy wrapper, and test the physical connection sometimes. 
                        // this is faster, when the testQuery would not otherwise be cached,
                        // and it avoids a potential statusOnException() double-check by the
                        // PooledConnection implementation should the test query provoke an
                        // Exception
                        Connection testConn;
                        if (scache != null) //when there is a statement cache...
                        {
                            // if it's the slow, default query, faster to test the raw Connection 
                            if (testQuery == null && connectionTesterIsDefault && c3p0PooledConnections)
                                testConn = ((AbstractC3P0PooledConnection) pc).getPhysicalConnection();
                            else //test will likely be faster on the proxied Connection, because the test query is probably cached
                                testConn = (proxyConn == null ? (openedConn = pc.getConnection()) : proxyConn); 
                        }
                        else //where there's no statement cache, better to use the physical connection, if we can get it
                        {
                            if (c3p0PooledConnections)
                                testConn = ((AbstractC3P0PooledConnection) pc).getPhysicalConnection();
                            else    
                                testConn = (proxyConn == null ? (openedConn = pc.getConnection()) : proxyConn);
                        }

                        if ( testQuery == null )
                            status = connectionTester.activeCheckConnection( testConn );
                        else
                        {
                            if (connectionTester instanceof UnifiedConnectionTester)
                            {
                                throwableHolder = thp.getThrowableHolder();
                                status = ((UnifiedConnectionTester) connectionTester).activeCheckConnection( testConn, testQuery, throwableHolder );
                            }
                            else if (connectionTester instanceof QueryConnectionTester)
                                status = ((QueryConnectionTester) connectionTester).activeCheckConnection( testConn, testQuery );
                            else
                            {
                                // System.err.println("[c3p0] WARNING: testQuery '" + testQuery +
                                // "' ignored. Please set a ConnectionTester that implements " +
                                // "com.mchange.v2.c3p0.advanced.QueryConnectionTester, or use the " +
                                // "DefaultConnectionTester, to test with the testQuery.");

                                logger.warning("[c3p0] testQuery '" + testQuery +
                                                "' ignored. Please set a ConnectionTester that implements " +
                                                "com.mchange.v2.c3p0.QueryConnectionTester, or use the " +
                                "DefaultConnectionTester, to test with the testQuery.");
                                status = connectionTester.activeCheckConnection( testConn );
                            }
                        }
                    }
                    catch (Exception e)
                    {
                        if (Debug.DEBUG)
                            logger.log(MLevel.FINE, "A Connection test failed with an Exception.", e);
                        //e.printStackTrace();
                        status = ConnectionTester.CONNECTION_IS_INVALID;
//                      System.err.println("rootCause ------>");
//                      e.printStackTrace();
                        rootCause = e;
                    }
                    finally
                    { 
                        if (rootCause == null)
                            rootCause = throwableHolder[0];
                        else if (throwableHolder[0] != null && logger.isLoggable(MLevel.FINE))
                            logger.log(MLevel.FINE, "Internal Connection Test Exception", throwableHolder[0]);
                        
                        if (throwableHolder != EMPTY_THROWABLE_HOLDER)
                            thp.returnThrowableHolder( throwableHolder );
                        
			//debug only
//  			if (openedConn != null)
//  			    new Exception("OPENEDCONN in testPooledConnection()").printStackTrace();

			// invalidate opened proxy connection
			// note that we close only what we might have opened in this method,
			// if we are handed a proxyConn by the client, we leave it for
			// that client to close()
                        ConnectionUtils.attemptClose( openedConn );

			// no! Connection should have been marked in use prior to test and should remain in use after
                        //unmarkPooledConnectionInUse( pc ); 
                    }

                    switch (status)
                    {
                    case ConnectionTester.CONNECTION_IS_OKAY:
                        break; //no problem, babe
                    case ConnectionTester.DATABASE_IS_INVALID:
                        rp.resetPool();
                        //intentional cascade...
                    case ConnectionTester.CONNECTION_IS_INVALID:
                        Exception throwMe;
                        if (rootCause == null)
                            throwMe = new SQLException("Connection is invalid");
                        else
                            throwMe = SqlUtils.toSQLException("Connection is invalid", rootCause);
                        throw throwMe;
                    default:
                        throw new Error("Bad Connection Tester (" + 
                                        connectionTester + ") " +
                                        "returned invalid status (" + status + ").");
                    }
                }

                public void destroyResource(Object resc, boolean checked_out) throws Exception
                { 
		    synchronized (inUseLockFetcher.getInUseLock(resc))
		    {
			try
			    {
				waitMarkPooledConnectionInUse((PooledConnection) resc);
                                
				if ( connectionCustomizer != null )
				    {
					Connection physicalConnection = null;
					try
					    { 
						physicalConnection =  ((AbstractC3P0PooledConnection) resc).getPhysicalConnection();
						
						connectionCustomizer.onDestroy( physicalConnection, parentDataSourceIdentityToken );
					    }
					catch (ClassCastException e)
					    {
						throw SqlUtils.toSQLException("Cannot use a ConnectionCustomizer with a non-c3p0 PooledConnection." +
									      " PooledConnection: " + resc + 
									      "; ConnectionPoolDataSource: " + cpds.getClass().getName(), e);
					    }
					catch (Exception e)
					    {
						if (logger.isLoggable( MLevel.WARNING ))
						    logger.log( MLevel.WARNING,
								"An exception occurred while executing the onDestroy() method of " + connectionCustomizer +
								". c3p0 will attempt to destroy the target Connection regardless, but this issue " +
								" should be investigated and fixed.",
								e );
					    }
				    }
				
				if (Debug.DEBUG && Debug.TRACE == Debug.TRACE_MAX && logger.isLoggable( MLevel.FINER ))
				    logger.log( MLevel.FINER, "Preparing to destroy PooledConnection: " + resc);
				
				if (c3p0PooledConnections)
				    ((AbstractC3P0PooledConnection) resc).closeMaybeCheckedOut( checked_out );
				else
				    ((PooledConnection) resc).close();
				
				// inaccurate, as Connections can be removed more than once
				//connectionCounter.decrement();
				
				if (Debug.DEBUG && Debug.TRACE == Debug.TRACE_MAX && logger.isLoggable( MLevel.FINER ))
				    logger.log( MLevel.FINER, 
						"Successfully destroyed PooledConnection: " + resc );
				//". Currently open Connections: " + connectionCounter.getValue() +
				//"; Failed close count: " + failedCloseCounter.getValue() +
				//"; Total processed by this pool: " + totalOpenedCounter.getValue());
			    }
			catch (Exception e)
			    {
				//failedCloseCounter.increment();
				
				if (Debug.DEBUG && Debug.TRACE == Debug.TRACE_MAX && logger.isLoggable( MLevel.FINER ))
				    logger.log( MLevel.FINER, "Failed to destroy PooledConnection: " + resc );
				//". Currently open Connections: " + connectionCounter.getValue() +
				//"; Failed close count: " + failedCloseCounter.getValue() +
				//"; Total processed by this pool: " + totalOpenedCounter.getValue());
				
				throw e;
			    }
			finally
			    { unmarkPooledConnectionInUse((PooledConnection) resc); }
		    }
		}
            }

            ResourcePool.Manager manager = new PooledConnectionResourcePoolManager();

            synchronized (fact)
            {
                fact.setMin( min );
                fact.setMax( max );
                fact.setStart( start );
                fact.setIncrement( inc );
                fact.setIdleResourceTestPeriod( idleConnectionTestPeriod * 1000);
                fact.setResourceMaxIdleTime( maxIdleTime * 1000 );
                fact.setExcessResourceMaxIdleTime( maxIdleTimeExcessConnections * 1000 );
                fact.setResourceMaxAge( maxConnectionAge * 1000 );
                fact.setExpirationEnforcementDelay( propertyCycle * 1000 );
                fact.setDestroyOverdueResourceTime( unreturnedConnectionTimeout * 1000 );
                fact.setDebugStoreCheckoutStackTrace( debugUnreturnedConnectionStackTraces );
                fact.setForceSynchronousCheckins( forceSynchronousCheckins );
                fact.setAcquisitionRetryAttempts( acq_retry_attempts );
                fact.setAcquisitionRetryDelay( acq_retry_delay );
                fact.setBreakOnAcquisitionFailure( break_after_acq_failure );
                rp = fact.createPool( manager );
            }
        }
        catch (ResourcePoolException e)
        { throw SqlUtils.toSQLException(e); }
    }

    public PooledConnection checkoutPooledConnection() throws SQLException
    { 
        //System.err.println(this + " -- CHECKOUT");
        try 
	    { 
		PooledConnection pc = (PooledConnection) this.checkoutAndMarkConnectionInUse(); 
		pc.addConnectionEventListener( cl );
		return pc;
	    }
        catch (TimeoutException e)
        { throw SqlUtils.toSQLException("An attempt by a client to checkout a Connection has timed out.", e); }
        catch (CannotAcquireResourceException e)
        { throw SqlUtils.toSQLException("Connections could not be acquired from the underlying database!", "08001", e); }
        catch (Exception e)
        { throw SqlUtils.toSQLException(e); }
    }
    
    private void waitMarkPhysicalConnectionInUse(Connection physicalConnection) throws InterruptedException
    {
        if (effectiveStatementCache)
            scache.waitMarkConnectionInUse(physicalConnection);
    }

    private boolean tryMarkPhysicalConnectionInUse(Connection physicalConnection)
    { return (effectiveStatementCache ? scache.tryMarkConnectionInUse(physicalConnection) : true); }
    
    private void unmarkPhysicalConnectionInUse(Connection physicalConnection)
    {
        if (effectiveStatementCache)
            scache.unmarkConnectionInUse(physicalConnection);
    }

    private void waitMarkPooledConnectionInUse(PooledConnection pooledCon) throws InterruptedException
    {
	if (c3p0PooledConnections)
	    waitMarkPhysicalConnectionInUse(((AbstractC3P0PooledConnection) pooledCon).getPhysicalConnection());
    }
    
    private boolean tryMarkPooledConnectionInUse(PooledConnection pooledCon)
    { 
	if (c3p0PooledConnections)
	    return tryMarkPhysicalConnectionInUse(((AbstractC3P0PooledConnection) pooledCon).getPhysicalConnection()); 
	else
	    return true;
    }
    
    private void unmarkPooledConnectionInUse(PooledConnection pooledCon)
    {
	if (c3p0PooledConnections)
	    unmarkPhysicalConnectionInUse(((AbstractC3P0PooledConnection) pooledCon).getPhysicalConnection());
    }

    private Boolean physicalConnectionInUse(Connection physicalConnection) throws InterruptedException
    {
        if (physicalConnection != null && effectiveStatementCache)
            return scache.inUse(physicalConnection);
	else
	    return null;
    }

    private Boolean pooledConnectionInUse(PooledConnection pc) throws InterruptedException
    {
        if (pc != null && effectiveStatementCache)
            return scache.inUse(((AbstractC3P0PooledConnection) pc).getPhysicalConnection());
	else
	    return null;
    }


     
    private Object checkoutAndMarkConnectionInUse() throws TimeoutException, CannotAcquireResourceException, ResourcePoolException, InterruptedException
    {
        Object out = null; 
	boolean success = false;
	while (! success)
	    {
		try
		    {
			out = rp.checkoutResource( checkoutTimeout );
			if (out instanceof AbstractC3P0PooledConnection)
			    {
				// cast should succeed, because effectiveStatementCache implies c3p0 pooled Connections
				AbstractC3P0PooledConnection acpc = (AbstractC3P0PooledConnection) out;
				Connection physicalConnection = acpc.getPhysicalConnection();
				success = tryMarkPhysicalConnectionInUse(physicalConnection);
			    }
			else
			    success = true; //we don't pool statements from non-c3p0 PooledConnections
		    }
		finally
		    {
			try { if (!success && out != null) rp.checkinResource( out );}
			catch (Exception e) { logger.log(MLevel.WARNING, "Failed to check in a Connection that was unusable due to pending Statement closes.", e); }
		    }
            }
        return out;
    }
    
    private void unmarkConnectionInUseAndCheckin(PooledConnection pcon) throws ResourcePoolException
    {
        if (effectiveStatementCache)
        {
            try
            {
                // cast should generally succeed, because effectiveStatementCache implies c3p0 pooled Connections
                // but clients can try to check-in whatever they want, so there are potential failures here
                AbstractC3P0PooledConnection acpc = (AbstractC3P0PooledConnection) pcon;
                Connection physicalConnection = acpc.getPhysicalConnection();
                unmarkPhysicalConnectionInUse(physicalConnection);
            }
            catch (ClassCastException e)
            {
                if (logger.isLoggable(MLevel.SEVERE))
                    logger.log(MLevel.SEVERE, 
                               "You are checking a non-c3p0 PooledConnection implementation into" +
                               "a c3p0 PooledConnectionPool instance that expects only c3p0-generated PooledConnections." +
                               "This isn't good, and may indicate a c3p0 bug, or an unusual (and unspported) use " +
                               "of the c3p0 library.", e);
            }
       }
       rp.checkinResource(pcon);
    }

    public void checkinPooledConnection(PooledConnection pcon) throws SQLException
    { 
        //System.err.println(this + " -- CHECKIN");
        try 
	    {
		pcon.removeConnectionEventListener( cl );
		unmarkConnectionInUseAndCheckin( pcon ); 
	    } 
        catch (ResourcePoolException e)
        { throw SqlUtils.toSQLException(e); }
    }

    public float getEffectivePropertyCycle() throws SQLException
    {
        try
        { return rp.getEffectiveExpirationEnforcementDelay() / 1000f; }
        catch (ResourcePoolException e)
        { throw SqlUtils.toSQLException(e); }
    }

    public int getNumThreadsAwaitingCheckout() throws SQLException
    {
        try
        { return rp.getNumCheckoutWaiters(); }
        catch (ResourcePoolException e)
        { throw SqlUtils.toSQLException(e); }
    }

    public int getStatementCacheNumStatements()
    { return scache == null ? 0 : scache.getNumStatements(); }

    public int getStatementCacheNumCheckedOut()
    { return scache == null ? 0 : scache.getNumStatementsCheckedOut(); }

    public int getStatementCacheNumConnectionsWithCachedStatements()
    { return scache == null ? 0 : scache.getNumConnectionsWithCachedStatements(); }

    public String dumpStatementCacheStatus()
    { return scache == null ? "Statement caching disabled." : scache.dumpStatementCacheStatus(); }

    public void close() throws SQLException
    { close( true ); }

    public void close( boolean close_outstanding_connections ) throws SQLException
    { 
        // System.err.println(this + " closing.");
        Exception throwMe = null;

        try { if (scache != null) scache.close(); }
        catch (SQLException e)
        { throwMe = e; }

        try 
        { rp.close( close_outstanding_connections ); }
        catch (ResourcePoolException e)
        {
            if ( throwMe != null && logger.isLoggable( MLevel.WARNING ) )
                logger.log( MLevel.WARNING, "An Exception occurred while closing the StatementCache.", throwMe);
            throwMe = e; 
        }

        if (throwMe != null)
            throw SqlUtils.toSQLException( throwMe );
    }

    class ConnectionEventListenerImpl implements ConnectionEventListener
    {

        //
        // We might want to check Connections in asynchronously, 
        // because this is called
        // (indirectly) from a sync'ed method of NewPooledConnection, but
        // NewPooledConnection may be closed synchronously from a sync'ed
        // method of the resource pool, leading to a deadlock. Checking
        // Connections in asynchronously breaks the cycle.
        //
        // But then we want checkins to happen quickly and reliably,
        // whereas pool shutdowns are rare, so perhaps it's best to
        // leave this synchronous, and let the closing of pooled
        // resources on pool closes happen asynchronously to break
        // the deadlock. 
        //
        // For now we're leaving both versions around, but with faster
        // and more reliable synchronous checkin enabled, and async closing
        // of resources in BasicResourcePool.close().
        //
        public void connectionClosed(final ConnectionEvent evt)
        { 
            //System.err.println("Checking in: " + evt.getSource());

            if (ASYNCHRONOUS_CONNECTION_EVENT_LISTENER) 
            {
                Runnable r = new Runnable()
                {
                    public void run()
                    { doCheckinResource( evt ); }
                };
                sharedTaskRunner.postRunnable( r );
            }
            else
                doCheckinResource( evt );
        }

        private void doCheckinResource(ConnectionEvent evt)
        {
            try
            { 
		//rp.checkinResource( evt.getSource() ); 
		checkinPooledConnection( (PooledConnection) evt.getSource() );
	    }
            catch (Exception e)
            { 
                //e.printStackTrace(); 
                logger.log( MLevel.WARNING, 
                                "An Exception occurred while trying to check a PooledConection into a ResourcePool.",
                                e );
            }
        }

        //
        // We might want to update the pool asynchronously, because this is called
        // (indirectly) from a sync'ed method of NewPooledConnection, but
        // NewPooledConnection may be closed synchronously from a sync'ed
        // method of the resource pool, leading to a deadlock. Updating
        // pool status asynchronously breaks the cycle.
        //
        // But then we want checkins to happen quickly and reliably,
        // whereas pool shutdowns are rare, so perhaps it's best to
        // leave all ConnectionEvent handling synchronous, and let the closing of pooled
        // resources on pool closes happen asynchronously to break
        // the deadlock. 
        //
        // For now we're leaving both versions around, but with faster
        // and more reliable synchrounous ConnectionEventHandling enabled, and async closing
        // of resources in BasicResourcePool.close().
        //
        public void connectionErrorOccurred(final ConnectionEvent evt)
        {
//          System.err.println("CONNECTION ERROR OCCURRED!");
//          System.err.println();
            if ( logger.isLoggable( MLevel.FINE ) )
                logger.fine("CONNECTION ERROR OCCURRED!");

            final PooledConnection pc = (PooledConnection) evt.getSource();
            int status;
            if (pc instanceof C3P0PooledConnection)
                status = ((C3P0PooledConnection) pc).getConnectionStatus();
            else if (pc instanceof NewPooledConnection)
                status = ((NewPooledConnection) pc).getConnectionStatus();
            else //default to invalid connection, but not invalid database
                status = ConnectionTester.CONNECTION_IS_INVALID;

            final int final_status = status;

            if (ASYNCHRONOUS_CONNECTION_EVENT_LISTENER) 
            {
                Runnable r = new Runnable()
                {
                    public void run()
                    { doMarkPoolStatus( pc, final_status ); }
                };
                sharedTaskRunner.postRunnable( r );
            }
            else
                doMarkPoolStatus( pc, final_status );
        }

        private void doMarkPoolStatus(PooledConnection pc, int status)
        {
            try
            {
                switch (status)
                {
                case ConnectionTester.CONNECTION_IS_OKAY:
                    throw new RuntimeException("connectionErrorOcccurred() should only be " +
                    "called for errors fatal to the Connection.");
                case ConnectionTester.CONNECTION_IS_INVALID:
                    rp.markBroken( pc );
                    break;
                case ConnectionTester.DATABASE_IS_INVALID:
                    if (logger.isLoggable(MLevel.WARNING))
                        logger.warning("A ConnectionTest has failed, reporting that all previously acquired Connections are likely invalid. " +
                        "The pool will be reset.");
                    rp.resetPool();
                    break;
                default:
                    throw new RuntimeException("Bad Connection Tester (" + connectionTester + ") " +
                                    "returned invalid status (" + status + ").");
                }
            }
            catch ( ResourcePoolException e )
            {
                //System.err.println("Uh oh... our resource pool is probably broken!");
                //e.printStackTrace();
                logger.log(MLevel.WARNING, "Uh oh... our resource pool is probably broken!", e);
            }
        }
    }

    public int getNumConnections() throws SQLException
    { 
        try { return rp.getPoolSize(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public int getNumIdleConnections() throws SQLException
    { 
        try { return rp.getAvailableCount(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public int getNumBusyConnections() throws SQLException
    { 
        try 
        { return rp.getAwaitingCheckinNotExcludedCount(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public int getNumUnclosedOrphanedConnections() throws SQLException
    {
        try { return rp.getExcludedCount(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }
    
    public long getStartTime() throws SQLException
    {
        try { return rp.getStartTime(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public long getUpTime() throws SQLException
    {
        try { return rp.getUpTime(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public long getNumFailedCheckins() throws SQLException
    {
        try { return rp.getNumFailedCheckins(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public long getNumFailedCheckouts() throws SQLException
    {
        try { return rp.getNumFailedCheckouts(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public long getNumFailedIdleTests() throws SQLException
    {
        try { return rp.getNumFailedIdleTests(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public Throwable getLastCheckinFailure() throws SQLException
    {
        try { return rp.getLastCheckinFailure(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public Throwable getLastCheckoutFailure() throws SQLException
    {
        try { return rp.getLastCheckoutFailure(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public Throwable getLastIdleTestFailure() throws SQLException
    {
        try { return rp.getLastIdleCheckFailure(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    public Throwable getLastConnectionTestFailure() throws SQLException
    {
        try { return rp.getLastResourceTestFailure(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }
    
    public Throwable getLastAcquisitionFailure() throws SQLException
    {
        try { return rp.getLastAcquisitionFailure(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    
Discards all Connections managed by the pool
and reacquires new Connections to populate.
Current checked out Connections will still
be valid, and should still be checked into the
pool (so the pool can destroy them).
/**
     * Discards all Connections managed by the pool
     * and reacquires new Connections to populate.
     * Current checked out Connections will still
     * be valid, and should still be checked into the
     * pool (so the pool can destroy them).
     */
    public void reset() throws SQLException
    { 
        try { rp.resetPool(); }
        catch ( Exception e )
        { 
            //e.printStackTrace();
            logger.log( MLevel.WARNING, null, e );
            throw SqlUtils.toSQLException( e );
        }
    }

    final static class ThrowableHolderPool
    {
        LinkedList l = new LinkedList();

        synchronized Throwable[] getThrowableHolder()
        {
            if (l.size() == 0)
                return new Throwable[1];
            else
                return (Throwable[]) l.remove(0);
        }

        synchronized void returnThrowableHolder(Throwable[] th)
        {
            th[0] = null;
            l.add(th);
        }
    }
    
}