/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2000-2009 Sun Microsystems, Inc. All rights reserved. 
 *
 * The contents of this file are subject to the terms of either the GNU
 * General Public License Version 2 only ("GPL") or the Common Development
 * and Distribution License ("CDDL") (collectively, the "License").  You may
 * not use this file except in compliance with the License.  You can obtain
 * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
 * or mq/legal/LICENSE.txt.  See the License for the specific language
 * governing permissions and limitations under the License.
 * 
 * When distributing the software, include this License Header Notice in each
 * file and include the License file at mq/legal/LICENSE.txt.  Sun designates
 * this particular file as subject to the "Classpath" exception as provided by
 * Sun in the GPL Version 2 section of the License file that accompanied this
 * code.  If applicable, add the following below the License Header, with the
 * fields enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 * 
 * Contributor(s):
 * 
 * If you wish your version of this file to be governed by only the CDDL or
 * only the GPL Version 2, indicate your decision by adding "[Contributor]
 * elects to include this software in this distribution under the [CDDL or GPL
 * Version 2] license."  If you don't indicate a single choice of license, a
 * recipient has the option to distribute your version of this file under
 * either the CDDL, the GPL Version 2 or  to extend the choice of license to
 * its licensees as provided above.  However, if you add GPL Version 2 code
 * and therefore, elected the GPL Version 2 license, then the option applies
 * only if the new code is made subject to such option by the copyright holder. 
 */

/*
 * @(#)DBConnectionPool.java	1.14 06/29/07
 */ 

package com.sun.messaging.jmq.jmsserver.persist.jdbc;

import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.config.ConfigListener;
import com.sun.messaging.jmq.jmsserver.config.PropertyUpdateException;
import com.sun.messaging.jmq.jmsserver.config.BrokerConfig;
import com.sun.messaging.jmq.jmsserver.resources.*;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;
import com.sun.messaging.jmq.jmsserver.FaultInjection;

import java.sql.*;
import java.util.*;
import java.util.Date;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * The DBConnection class represents a pool of connections to one database.
 */
public class DBConnectionPool implements DBConstants {

    private static boolean DEBUG = false;

    public static final String REAP_INTERVAL_PROP =
        DBManager.JDBC_PROP_PREFIX + ".connection.reaptime";
    public static final int DEFAULT_REAP_INTERVAL = 300; // secs

    public static final String NUM_CONN_PROP =
        DBManager.JDBC_PROP_PREFIX + ".connection.limit";
    public static final String MIN_CONN_PROP =
        DBManager.JDBC_PROP_PREFIX + ".min_connections";
    public static final String MAX_CONN_PROP =
        DBManager.JDBC_PROP_PREFIX + ".max_connections";
    static final int DEFAULT_NUM_CONN = 5;

    private static int minConnections;
    private static int maxConnections;

    private static boolean initialized = false;
    private static ReentrantLock lock = new ReentrantLock();

    private static LinkedBlockingQueue idleConnections = new LinkedBlockingQueue();
    private static ConcurrentHashMap activeConnections = new ConcurrentHashMap();
    private static ConnectionReaperTask connectionReaper = null;
    private static long reapInterval;

    private static DBManager dbmgr = null;
    private static Logger logger = Globals.getLogger();
    private static BrokerResources br = Globals.getBrokerResources();

    private static ConfigListener cfgListener = new ConfigListener() {
        public void validate(String name, String value)
            throws PropertyUpdateException {
            if (name.equals(MIN_CONN_PROP)) {
                int min = 0;
                try {
                    min = Integer.parseInt(value);
                } catch (Exception e) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting, br.getString(
                        BrokerResources.X_BAD_PROPERTY_VALUE, name + "=" + value), e);
                }
                if (min < 1) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting,
                        "A minimum value of 1 connection is required");
                } else if (min > maxConnections) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting,
                        "Minimum connections " + min +
                        " is greater than maximum connections " + maxConnections);
                }
            } else if (name.equals(MAX_CONN_PROP)) {
                int max = 0;
                try {
                    max = Integer.parseInt(value);
                } catch (Exception e) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting, br.getString(
                        BrokerResources.X_BAD_PROPERTY_VALUE, name + "=" + value), e);
                }
                if (max < minConnections) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting,
                        "Maximum connections " + max +
                        " is less than minimum connections " + minConnections);
                }
            } else if (name.equals(REAP_INTERVAL_PROP)) {
                int reaptime = 0;
                try {
                    reaptime = Integer.parseInt(value);
                } catch (Exception e) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting, br.getString(
                        BrokerResources.X_BAD_PROPERTY_VALUE, name + "=" + value), e);
                }
                if (reaptime < 60) {
                    throw new PropertyUpdateException(
                        PropertyUpdateException.InvalidSetting,
                        "A minimum value of 60 seconds is required for reap time interval");
                }
            }
        }

        public boolean update(String name, String value) {
            BrokerConfig cfg = Globals.getConfig();

            lock.lock();
            try {
                if (name.equals(MAX_CONN_PROP)) {
                    maxConnections = cfg.getIntProperty(MAX_CONN_PROP);
                } else if (name.equals(MIN_CONN_PROP)) {
                    minConnections = cfg.getIntProperty(MIN_CONN_PROP);
                } else if (name.equals(REAP_INTERVAL_PROP)) {
                    reapInterval = cfg.getLongProperty(REAP_INTERVAL_PROP) * 1000L;
                }
            } finally {
                lock.unlock();
            }

            // Start connection reaper to remove excess connections
            if (maxConnections > minConnections) {
                if (connectionReaper != null) {
                    connectionReaper.cancel(); // Cancel the old reaper task
                }

                connectionReaper = new ConnectionReaperTask();
                Globals.getTimer().schedule(
                    connectionReaper, reapInterval, reapInterval);
            }

            return true;
        }
    };

    /**
     * Establish a pool of database connections.
     */
    static void init(DBManager mgr) throws BrokerException {

        if ( !initialized ) {
            lock.lock();
            try {
                if ( initialized ) {
                    return;
                }

                dbmgr = mgr;

                // Check deprecated "imq.persist.jdbc.connection.limit" property
                int numConnections = Globals.getConfig().getIntProperty(
                    NUM_CONN_PROP, DEFAULT_NUM_CONN);

                if (numConnections < 1) {
                    numConnections = DEFAULT_NUM_CONN;
                    logger.log(Logger.WARNING,
                        "Invalid number of connections specified, set to default of " +
                        numConnections);
                }

                minConnections = Globals.getConfig().getIntProperty(
                    MIN_CONN_PROP, numConnections);

                if (minConnections < 1) {
                    minConnections = numConnections;
                    logger.log(Logger.WARNING,
                        "Invalid number of minimum connections specified, set to default of " +
                        minConnections);
                }

                maxConnections = Globals.getConfig().getIntProperty(
                    MAX_CONN_PROP, numConnections);

                if (maxConnections < minConnections) {
                    maxConnections = minConnections;
                    logger.log(Logger.WARNING,
                        "Invalid number of maximum connections specified, set to default of " +
                        maxConnections);
                }

                long reapTime = Globals.getConfig().getLongProperty(
                                REAP_INTERVAL_PROP, DEFAULT_REAP_INTERVAL);

                if (reapTime < 60) {
                    reapTime = DEFAULT_REAP_INTERVAL;
                    logger.log(Logger.WARNING,
                        "Invalid reap time interval for pool maintenance thread specified, set to default of " +
                        reapTime);
                }
                
                reapInterval = reapTime * 1000L;

                // With embedded DB, if autocreate store is enabled then we
                // need to create the DB now; otherwise we run into a chicken and
                // egg problem because we will not be able to create a connection
                // to check if the store exists.               
                if (dbmgr.getCreateDBURL() != null &&
                    Globals.getConfig().getBooleanProperty(Store.CREATE_STORE_PROP, false)) {
                    try {
                        Connection conn = dbmgr.connectToCreate();
                        conn.close();
                    } catch (Exception e) {
                        String url = dbmgr.getCreateDBURL();
                        logger.log(Logger.ERROR, BrokerResources.E_CREATE_DATABASE_TABLE_FAILED, url, e);
                        throw new BrokerException(br.getString(
                            BrokerResources.E_CREATE_DATABASE_TABLE_FAILED, url, e));
                    }
                }

                for (int i = 0; i < minConnections; i++) {
                    Connection conn = dbmgr.newConnection(true);
                    idleConnections.offer(conn);
                }

                // Registerd listener so we can dynamically changed pool value
                Globals.getConfig().addListener(MIN_CONN_PROP, cfgListener);
                Globals.getConfig().addListener(MAX_CONN_PROP, cfgListener);
                Globals.getConfig().addListener(REAP_INTERVAL_PROP, cfgListener);

                // Start connection reaper to remove excess connections
                if (maxConnections > minConnections) {
                    if (connectionReaper != null) {
                        connectionReaper.cancel(); // Cancel the old reaper task
                    }

                    connectionReaper = new ConnectionReaperTask();
                    Globals.getTimer().schedule(connectionReaper, reapInterval, reapInterval);
                }

                initialized = true;
            } finally {
                lock.unlock();
            }
        }
    }

    /**
     * Closes all available connections. Should be called when the broker is
     * shutting down and all store operations are done so there should not
     * be any connection in the activeConnections list.
     */
    static void close() {

        if (!initialized) {
	    return;
        }

        lock.lock();
        try {
            if (connectionReaper != null) {
                connectionReaper.cancel();
                connectionReaper = null;
            }

            Globals.getConfig().removeListener(MIN_CONN_PROP, cfgListener);
            Globals.getConfig().removeListener(MAX_CONN_PROP, cfgListener);
            Globals.getConfig().removeListener(REAP_INTERVAL_PROP, cfgListener);

            // Close all connections
            Iterator itr = idleConnections.iterator();
            while (itr.hasNext()) {
                Connection c = (Connection)itr.next();
                try {
                    c.close();
                } catch (SQLException e) {
                    logger.log(Logger.WARNING, BrokerResources.X_CLOSE_DATABASE_FAILED, e);
                }
            }

            idleConnections.clear();

            initialized = false;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Recreates all the connections.
     * 
     * Should be used to remove stale connections after the DB is restarted.
     */
    static void reset() throws BrokerException {
        
        if (Store.getDEBUG()) {
	    logger.log(Logger.DEBUG,
		"DBConnectionPool.reset(): reset connection pool");
        }
                
        if (!initialized) {
	    return;
        }
        
        Collection oldConnections = new ArrayList(maxConnections);
        
        lock.lock();
        try {                       
            activeConnections.clear();
            idleConnections.drainTo(oldConnections); 
            
             // Recreates the connections
            for (int i = 0; i < minConnections; i++) {
                Connection conn = dbmgr.newConnection(true);
                idleConnections.offer(conn);
            }
            
            // Now, close the old connections
            Iterator itr = oldConnections.iterator();
            while (itr.hasNext()) {
                Connection conn = (Connection)itr.next();
                try {
                    conn.close();
                } catch (SQLException e) {
                    // Ignored error
                }
            }
        } finally {
            lock.unlock();
        }        
    }

    /**
     * Checks out a connection from the pool.
     * @throws BrokerException
     */
    static Connection getConnection() throws BrokerException {

        if (Store.getDEBUG() || DEBUG) {
            logger.log(Logger.INFO, "["+Thread.currentThread()+"]DBConnectionPool.getConnection()["+
                                     idleConnections.size()+", "+activeConnections.size()+"]");
        }

        if (DEBUG) {
            FaultInjection fi = FaultInjection.getInjection();
            if (fi.FAULT_INJECTION) {
                fi.checkFaultAndSleep(FaultInjection.FAULT_JDBC_GETCONN_1, null);
            }
        }
        
        boolean createdNew = false, pollWait = false;
        Connection conn = (Connection)idleConnections.poll();
        if (conn == null && (activeConnections.size() < maxConnections)) {
            conn = dbmgr.newConnection(true);
            if (Store.getDEBUG() || DEBUG) {
                createdNew = true;
            }
        } else {
            // Wait until a connetion is free up
            while (conn == null) {
                try {
                    if (Store.getDEBUG() || DEBUG) {
                        if (!pollWait) pollWait = true;
                    }
                    conn = (Connection)idleConnections.poll(60, TimeUnit.SECONDS);
                    if (conn == null) {
                        // Something is wrong! print the stack trace
                        StringBuffer buff = new StringBuffer(1024);
                        Iterator itr = activeConnections.entrySet().iterator();
                        while (itr.hasNext()) {
                            Map.Entry e = (Map.Entry)itr.next();
                            Thread t = (Thread)e.getValue();
                            buff.append("\n")
                                .append(t.getName())
                                .append(": using connection: ")
                                .append(e.getKey());
                            StackTraceElement[] trace = t.getStackTrace();
                            for (int i=0; i < trace.length; i++) {
                                buff.append("\n\tat " + trace[i]);
                            }
                        }
                        logger.log(Logger.INFO,
                            "Unable to obtain database connection after waiting for 60 secs [pool size: min="+
                            minConnections + ", max="+ maxConnections +
                            "]: stack trace of connection pool:" + buff.toString());
                    }
                } catch (Exception e) {
                    logger.log(Logger.DEBUG,
                        "DBConnectionPool.getConnection(): " + e);
                }
            }

            // Check if connection is closed
            boolean isClosed = false;
            try {
				isClosed = conn.isClosed();
			} catch (SQLException e) {
				try {
					// something wrong with connection so try and close explicitly
					// and mark as closed
					  logger.log(Logger.INFO, "exception calling isClosed on connection. Will close");
					conn.close();
				} catch (SQLException e2) {
				}
				isClosed = true;
			}

            if ( isClosed ) {
                // if we have an invalid connection, reconnect
                try {
                    Connection newconn = dbmgr.newConnection(true);
                    conn = newconn;

                    logger.log(Logger.INFO, br.getString(
                        BrokerResources.I_RECONNECT_TO_DB,
                        dbmgr.getOpenDBURL()));
                } catch (BrokerException e) {
                    logger.log(Logger.ERROR, br.getString(
                        BrokerResources.X_RECONNECT_TO_DB_FAILED,
                        dbmgr.getOpenDBURL()), e);

                    // just stick the bad connection back to the idle list
                    // so that we will try to reconnect again
                    idleConnections.add(conn);
                    throw e;
                }
            }
        }

        // move the connection in the activeConnections list
        Thread borrower = Thread.currentThread();
        activeConnections.put(conn, borrower);

        if (Store.getDEBUG() || DEBUG) {
            logger.log(Logger.INFO, "DBConnectionPool.getConnection()["+createdNew+","+pollWait+"]: " +
                borrower.getName() + " [" + new Date() +
                "]: check out connection: " + conn);
        }

        return conn;
    }

    /**
     * Checks in a connection to the pool.
     */
    static void freeConnection(Connection conn) {

        if (Store.getDEBUG()) {
	    logger.log(Logger.DEBUG,
		"DBConnectionPool.freeConnection(): check in connection: " + conn);
        }

        if (activeConnections.remove(conn) == null) {
            // Just free the connection that does not come from the pool!
            try {
                conn.close();
            } catch (SQLException e) {
                logger.logStack(Logger.WARNING,
                    Globals.getBrokerResources().getKString(
                        BrokerResources.E_INTERNAL_BROKER_ERROR,
                        "Unable to close JDBC resources", e), e);
            }
        } else {
            // Return the connection back to the pool
            idleConnections.offer(conn);
        }
    }

    static void reapExcessConnection() {

        int idleCnt = idleConnections.size();
        int activeCnt = activeConnections.size();

        if (Store.getDEBUG()) {
            logger.log(Logger.DEBUG,
                "DBConnectionPool.reapExcessConnection(): pool size: min=" +
                minConnections + ", max=" + maxConnections + ", active=" +
                activeCnt + ", idle=" + idleCnt);
        }

        while (idleCnt > 0 && (activeCnt + idleCnt) > minConnections) {
            // Free dynamically created connections to maintain pool size
            try {
                // Removed excess connection from the pool
                Connection conn = (Connection)idleConnections.poll();
                if (conn != null) {
                    conn.close();
                }                
            } catch (SQLException e) {
                logger.logStack(Logger.WARNING,
                    Globals.getBrokerResources().getKString(
                        BrokerResources.E_INTERNAL_BROKER_ERROR,
                        "Unable to close JDBC resources", e), e);
            }
            idleCnt = idleConnections.size();
            activeCnt = activeConnections.size();
        }
    }

    /**
     * The validation check runs a simple SQL query over the connection
     * to see if it throws an exception.
     */
    static boolean validate( Connection conn ) {

        try {
            conn.getMetaData();
        } catch (Exception e) {
            logger.log(Logger.DEBUG,
                "DBConnectionPool.validate(): Lost database connection to "
                + dbmgr.getOpenDBURL(), e);
            return false;
        }
        return true;
    }

    // returning true indicates that the caller should retry the operation
    // will return false if the connection is still good
    static boolean handleException(Connection conn, Throwable t) {

        // test to see if the connection is good
        if ( validate( conn ) ) {

            logger.log(Logger.DEBUG,
                "connection is good; return false (no need to retry)");

            // successful so connection is ok
            return false; // no retry
        }

        return true;
    }

    static class ConnectionReaperTask extends TimerTask
    {
        private volatile boolean canceled = false;

        public boolean cancel() {
            canceled = true;
            return super.cancel();
        }

        public void run() {
            if (canceled) {
                return;
            }

            try {
                reapExcessConnection();
            } catch (Exception e) {
                Globals.getLogger().logStack( Logger.ERROR,
                    BrokerResources.E_INACTIVE_SESSION_REMOVAL_FAILED, e );
            }
        }
    }
}
