/*
 * 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. 
 */

/*
 * @(#)DBManager.java	1.72 06/29/07
 */ 

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

import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.util.StringUtil;
import com.sun.messaging.jmq.util.Password;
import com.sun.messaging.jmq.jmsserver.util.*;
import com.sun.messaging.jmq.jmsserver.config.*;
import com.sun.messaging.jmq.jmsserver.*;
import com.sun.messaging.jmq.jmsserver.cluster.BrokerState;
import com.sun.messaging.jmq.jmsserver.resources.*;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.persist.HABrokerInfo;
import com.sun.messaging.jmq.io.Status;

import java.io.*;
import java.sql.*;
import java.util.*;
import java.net.*;
import java.lang.reflect.*;
import javax.sql.*;

/**
 * This class holds database related properties and includes
 * methods to get database connections.
 */
public final class DBManager implements DBConstants {

    /**
     * properties used:
     *
     * - vendor specific jdbc drvier
     * jmq.persist.dbVendor=<Database vendor>
     *
     * - vendor specific jdbc drvier
     * jmq.persist.jdbc.<dbVendor>.driver=<jdbcdriver class>
     * jmq.persist.jdbc.driver=<jdbcdriver class>
     *
     * - vendor specific database url to get a database connection
     * jmq.persist.jdbc.<dbVendor>.opendburl=<url to open database>
     * jmq.persist.jdbc.opendburl=<url to open database>
     *
     * - vendor specific database url to get a database connection to create
     *   a database
     * jmq.persist.jdbc.<dbVendor>.createdburl=<url to create database>
     * jmq.persist.jdbc.createdburl=<url to create database>
     *
     * - vendor specific database url to shutdown the connection (optional)
     * jmq.persist.jdbc.closedburl=<url to close database connection>
     * jmq.persist.jdbc.<dbVendor>.closedburl=<url to close database connection>
     *
     * - user name used to open database connection (optional)
     * jmq.persist.jdbc.<dbVendor>.user=<username>
     * jmq.persist.jdbc.user=<username>
     *
     * - password used to open database connection (private)
     * jmq.persist.jdbc.<dbVendor>.password=<password>
     * jmq.persist.jdbc.password=<password>
     *
     * - password used to open database connection (optional)
     * jmq.persist.jdbc.<dbVendor>.needpassword=[true|false]
     * jmq.persist.jdbc.needpassword=[true|false]
     *
     * - brokerid to make table names unique per broker instance
     * jmq.brokerid=<alphanumeric id>
     *
     * - clusterid to make table names unique for shared store per HA cluster
     * imq.cluster.clusterid=<alphanumeric id>
     */
    static final String STORE_TYPE_PROP = Globals.IMQ + ".persist.store";
    static final String JDBC_STORE_TYPE = "jdbc";

    static final String JDBC_PROP_PREFIX = Globals.IMQ + ".persist.jdbc";
    static final String FALLBACK_USER_PROP = JDBC_PROP_PREFIX + ".user";
    static final String FALLBACK_PWD_PROP = JDBC_PROP_PREFIX + ".password";

    static final String TRANSACTION_RETRY_MAX_PROP =
        JDBC_PROP_PREFIX + ".transaction.retry.max";
    static final String TRANSACTION_RETRY_DELAY_PROP =
        JDBC_PROP_PREFIX + "transaction.retry.delay";

    static final int TRANSACTION_RETRY_MAX_DEFAULT = 5;
    static final long TRANSACTION_RETRY_DELAY_DEFAULT = 2000;

    static final String BROKERID_PROP = Globals.IMQ + ".brokerid";
    static final String TABLE_PROP_PREFIX = JDBC_PROP_PREFIX + ".table";

    static final String UNKNOWN_VENDOR = "unknown";
    static final boolean DEFAULT_NEEDPASSWORD = false;

    String vendor = null;
    String vendorPropPrefix = null;
    String tablePropPrefix = null;
    String vendorProp = null;
    String driverProp = null;
    String openDBUrlProp = null;
    String createDBUrlProp = null;
    String closeDBUrlProp = null;
    String userProp = null;
    String passwordProp = null;
    String needPasswordProp = null;

    int txnRetryMax;    // Max number of retry
    long txnRetryDelay; // Number of milliseconds to wait between retry

    // database connection
    private String driver = null;
    private String openDBUrl = null;
    private String createDBUrl = null;
    private String closeDBUrl = null;
    private String user = null;
    private String password = null;
    private boolean needPassword = true;
    private boolean isDataSource = false;
    private boolean isPoolDataSource = false;
    private Object dataSource = null;

    private boolean isHADB = false;
    private boolean isOracle = false;
    private boolean isOraDriver = false;
    private boolean isMysql = false;
    private boolean supportBatch = false;
    private String dbProductName = null;
    private String dbProductVersion = null;

    private String tableSuffix = null;

    // cluster id to make table names unique per cluster
    private String clusterID = null;

    // broker id to make table names unique per broker
    private String brokerID = null;

    // DAO factory
    private DAOFactory daoFactory = null;

    private BrokerConfig config = Globals.getConfig();
    private BrokerResources br = Globals.getBrokerResources();
    private Logger logger = Globals.getLogger();

    private static final Object lock = DBManager.class;
    private static DBManager dbMgr = null;

    // HashMap of database tables. table name->TableSchema
    private HashMap tableSchemas = new HashMap();

    // array of table names used in version 370 store
    private static String v370tableNames[] = {
                                         VERSION_TBL_37,
                                         CONFIGRECORD_TBL_37,
                                         DESTINATION_TBL_37,
                                         INTEREST_TBL_37,
                                         MESSAGE_TBL_37,
                                         PROPERTY_TBL_37,
                                         INTEREST_STATE_TBL_37,
                                         TXN_TBL_37,
                                         TXNACK_TBL_37
    };

    // array of table names used in version 350 store
    private static String v350tableNames[] = {
                                         VERSION_TBL_35,
                                         CONFIGRECORD_TBL_35,
                                         DESTINATION_TBL_35,
                                         INTEREST_TBL_35,
                                         MESSAGE_TBL_35,
                                         PROPERTY_TBL_35,
                                         INTEREST_STATE_TBL_35,
                                         TXN_TBL_35,
                                         TXNACK_TBL_35
    };

    /**
     * Get DBManager method for singleton pattern.
     * @return DBManager
     * @throws BrokerException
     */
    public static DBManager getDBManager() throws BrokerException {
        if (dbMgr == null) {
            synchronized(lock) {
                if (dbMgr == null) {
                    dbMgr = new DBManager();
                    
                    // get table schema
                    dbMgr.loadTableSchema();
                }

                // Initialize connection pool after DBManager is initialized!
                DBConnectionPool.init(dbMgr);

                // Initialize relevant infos
                int maxTableNameLength = 0;
                Connection conn = null;
                try {
                    conn = dbMgr.getConnection( true );

                    DatabaseMetaData dbMetaData = conn.getMetaData();
                    dbMgr.dbProductName = dbMetaData.getDatabaseProductName();
                    dbMgr.dbProductVersion = dbMetaData.getDatabaseProductVersion();
                    dbMgr.supportBatch = dbMetaData.supportsBatchUpdates();
                    maxTableNameLength = dbMetaData.getMaxTableNameLength();
                    String driverv = dbMetaData.getDriverVersion();

                    // Check to see if we're using an Oracle driver so
                    // we know how to deal w/ Oracle LOB handling!
                    dbMgr.isOraDriver = "oracle".equalsIgnoreCase(dbMgr.dbProductName);

                    String logMsg = new StringBuffer(256)
                        .append("DBManager: database product name=")
                        .append(dbMgr.dbProductName)
                        .append(", database version number=")
                        .append(dbMgr.dbProductVersion)
                        .append(", driver version number=")
                        .append(driverv)
                        .append(", supports batch updates=")
                        .append(dbMgr.supportBatch)
                        .toString();
                    dbMgr.logger.log(Logger.FORCE, dbMgr.dbProductName+", "+dbMgr.dbProductVersion+", "+driverv);
                    dbMgr.logger.log((Store.getDEBUG()? Logger.INFO:Logger.DEBUG), logMsg);
                } catch (SQLException e) {
                    dbMgr.logger.log(Logger.WARNING,
                        BrokerResources.X_GET_METADATA_FAILED, e);
                } finally {
                    Util.close( null, null, conn );
                }

                if (maxTableNameLength > 0) {
                    // We do know the max number of chars allowed for a table
                    // name so verify brokerID or clusterID is within limit.
                    if (Globals.getHAEnabled()) {
                        if ((dbMgr.clusterID.length() + 14) > maxTableNameLength) {
                            throw new BrokerException(dbMgr.br.getKString(
                                BrokerResources.E_CLUSTER_ID_TOO_LONG, dbMgr.clusterID, Integer.valueOf(14)));
                        }
                    } else {
                        if ((dbMgr.brokerID.length() + 14) > maxTableNameLength) {
                            throw new BrokerException(dbMgr.br.getKString(
                                BrokerResources.E_BROKER_ID_TOO_LONG, dbMgr.brokerID, Integer.valueOf(14)));
                        }
                    }
                }
            }
        }
        return dbMgr;
    }

    /**
     * When instantiated, the object configures itself by reading the
     * properties specified in BrokerConfig.
     */
    private DBManager() throws BrokerException {

        txnRetryMax = config.getIntProperty( TRANSACTION_RETRY_MAX_PROP,
            TRANSACTION_RETRY_MAX_DEFAULT );
        txnRetryDelay = config.getLongProperty( TRANSACTION_RETRY_DELAY_PROP,
            TRANSACTION_RETRY_DELAY_DEFAULT);

        // get store type and double check that 'jdbc' is specified
        String type = config.getProperty(STORE_TYPE_PROP);
        if (type == null || !type.equals(JDBC_STORE_TYPE)) {
            type = (type == null) ? "" : type;
            throw new BrokerException(
                br.getKString(BrokerResources.E_NOT_JDBC_STORE_TYPE, type));
        }

        vendorProp = JDBC_PROP_PREFIX + ".dbVendor";
        vendor = config.getProperty(vendorProp, UNKNOWN_VENDOR);
        vendorPropPrefix = JDBC_PROP_PREFIX + "." + vendor;

        tablePropPrefix = vendorPropPrefix + ".table";

        // get jdbc driver property
        driverProp = vendorPropPrefix + ".driver";
        driver = config.getProperty(driverProp);
        if (driver == null || (driver.length() == 0)) {
            // try fallback prop
            String fallbackProp = JDBC_PROP_PREFIX + ".driver";
            driver = config.getProperty(fallbackProp);
            if (driver == null || (driver.length() == 0)) {
                throw new BrokerException(
                    br.getKString(BrokerResources.E_NO_JDBC_DRIVER_PROP, driverProp));
            }
            driverProp = fallbackProp;
        }
        logger.log(logger.FORCE, driverProp+"="+driver);

        // get open database url property (optional for DataSource)
        openDBUrlProp = vendorPropPrefix + ".opendburl";
        openDBUrl = config.getProperty(openDBUrlProp);
        if (openDBUrl == null || (openDBUrl.length() == 0)) {
            // try fallback prop
            String fallbackProp = JDBC_PROP_PREFIX + ".opendburl";
            openDBUrl = config.getProperty(fallbackProp);
            openDBUrlProp = fallbackProp;
        }
        openDBUrl = StringUtil.expandVariables(openDBUrl, config);
        if (openDBUrl != null) {
            logger.log(logger.FORCE, openDBUrlProp+"="+openDBUrl);
        }

        //
        // optional properties
        //

        // get create database url property
        createDBUrlProp = vendorPropPrefix + ".createdburl";
        createDBUrl = config.getProperty(createDBUrlProp);
        if (createDBUrl == null || (createDBUrl.length() == 0)) {
            // try fallback prop
            String fallbackProp = JDBC_PROP_PREFIX + ".createdburl";
            createDBUrl = config.getProperty(fallbackProp);
            if (createDBUrl != null) {
                createDBUrlProp = fallbackProp;
            }
        }
        createDBUrl = StringUtil.expandVariables(createDBUrl, config);
        if (createDBUrl != null) {
            logger.log(logger.FORCE, createDBUrlProp+"="+createDBUrl);
        }

        // get url to shutdown database
        closeDBUrlProp = vendorPropPrefix + ".closedburl";
        closeDBUrl = config.getProperty(closeDBUrlProp);
        if (closeDBUrl == null || (closeDBUrl.length() == 0)) {
            // try fallback prop
            String fallbackProp = JDBC_PROP_PREFIX + ".closedburl";
            closeDBUrl = config.getProperty(fallbackProp);
            if (closeDBUrl != null) {
                closeDBUrlProp = fallbackProp;
            }
        }
        closeDBUrl = StringUtil.expandVariables(closeDBUrl, config);
        if (closeDBUrl != null) {
            logger.log(logger.FORCE, closeDBUrlProp+"="+closeDBUrl);
        }

        // user name to open connection
        userProp = vendorPropPrefix + ".user";
        user = config.getProperty(userProp);
        if (user == null) {
            // try fallback prop
            user = config.getProperty(FALLBACK_USER_PROP);
            if (user != null) {
                userProp = FALLBACK_USER_PROP;
            }
        }

        // broker id
        brokerID = Globals.getBrokerID();
        if (brokerID == null || brokerID.length() == 0 ||
            !Util.isAlphanumericString(brokerID)) {
            throw new BrokerException(
                br.getKString(BrokerResources.E_BAD_BROKER_ID, brokerID));
        }

        // table suffix
        if (Globals.getHAEnabled()) {
            clusterID = Globals.getClusterID();
            if (clusterID == null || clusterID.length() == 0 ||
                !Util.isAlphanumericString(clusterID)) {
                    throw new BrokerException(br.getKString(
                        BrokerResources.E_BAD_CLUSTER_ID, clusterID));
            }

            // Use cluster ID as the suffix
            tableSuffix = "C" + clusterID;
        } else {
            // Use broker ID as the suffix
            tableSuffix = "S" + brokerID;
        }

        // Fix version 370 & 350 table names
        fixOldTableNames();

        // load jdbc driver
        try {
            Class driverCls = Class.forName(driver);
            Object driverObj = driverCls.newInstance();
            // Check if driver is a DataSource
            if (driverObj instanceof ConnectionPoolDataSource) {
                isDataSource = true;
                isPoolDataSource = true;
            } else if (driverObj instanceof DataSource) {
                isDataSource = true;
            } else {
                // Not using DataSource make sure driver's url is specified
                if (openDBUrl == null || (openDBUrl.length() == 0)) {
                    throw new BrokerException(br.getKString(
                        BrokerResources.E_NO_DATABASE_URL_PROP, openDBUrlProp));
                }
            }

            // Initialize DataSource properties
            if ( isDataSource ) {
                dataSource = driverObj;
                initDataSource(driverCls, dataSource);
            }
        } catch (InstantiationException e) {
            throw new BrokerException(
                br.getKString(BrokerResources.E_CANNOT_LOAD_JDBC_DRIVER, driver), e);
        } catch (IllegalAccessException e) {
            throw new BrokerException(
                br.getKString(BrokerResources.E_CANNOT_LOAD_JDBC_DRIVER, driver), e);
        } catch (ClassNotFoundException e) {
            throw new BrokerException(
                br.getKString(BrokerResources.E_CANNOT_LOAD_JDBC_DRIVER, driver), e);
        }

        // password to open connection; do this last because we want to init
        // the datasource to get the url that might be needed in getPassword()
        password = getPassword();

        // verify dbVendor is not unknown, i.e. upgrading from old store
        if (UNKNOWN_VENDOR.equals(vendor)) {
            // need to figure out the vendor
            Connection conn = null;
            String dbName = null;
            try {
                conn = newConnection(true);
                DatabaseMetaData dbMetaData = conn.getMetaData();
                dbName = dbMetaData.getDatabaseProductName();
            } catch (Exception e) {
                // Ignore error for now!
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (SQLException e) {}
                }
            }

            if ("oracle".equalsIgnoreCase(dbName)) {
                vendor = "oracle";
                vendorPropPrefix = JDBC_PROP_PREFIX + "." + vendor;
                tablePropPrefix = vendorPropPrefix + ".table";
            } else {
                throw new BrokerException(
                    br.getKString(BrokerResources.E_NO_DATABASE_VENDOR_PROP));
            }
        }

        if ( vendor.equalsIgnoreCase( "hadb" ) ) {
            isHADB = true;
            checkHADBJDBCLogging();
        } else if ( vendor.equalsIgnoreCase( "oracle" ) ) {
            isOracle = true;
        } else if ( vendor.equalsIgnoreCase( "mysql" ) ) {
            isMysql = true;
        }
    }

    /**
     * Get a database connection using the CREATEDB_URL.
     */
    Connection connectToCreate() throws BrokerException {

        if (createDBUrl == null) {
            throw new BrokerException(br.getKString(
                BrokerResources.E_NO_DATABASE_URL_PROP, createDBUrlProp));
        }

        // make database connection
        Connection conn = null;
        try {
            if (user == null) {
                conn = DriverManager.getConnection(createDBUrl);
            } else {
                conn = DriverManager.getConnection(createDBUrl, user, password);
            }

            conn.setAutoCommit(false);
        } catch (SQLException e) {
            if (conn != null) {
                try {
                    conn.close();
                } catch(SQLException sqe) {
                    logger.logStack(Logger.WARNING,
                        Globals.getBrokerResources().getKString(
                            BrokerResources.E_INTERNAL_BROKER_ERROR,
                            "Unable to close JDBC resources", e), e);
                }
            }
            throw new BrokerException(br.getKString(
                BrokerResources.E_CANNOT_GET_DB_CONNECTION, createDBUrl), e);
        }

        return conn;
    }

    /**
     * Get a connection from the connection pool.
     */
    public Connection getConnection(boolean autocommit)
        throws BrokerException {

        boolean gotException = false;
        Connection conn = DBConnectionPool.getConnection();
        try {
            // Set connection behavior

            try {
                if (conn.getAutoCommit() != autocommit) {
                    conn.setAutoCommit(autocommit);
                }
            } catch ( SQLException e ) {
                gotException = true;
                throw new BrokerException(
                    br.getKString(BrokerResources.X_INTERNAL_EXCEPTION,
                    "Unable to set connection's auto-commit mode"), e);
            }

            int txnIsolation = -1;
            try {
                if (isHADB) {
                    txnIsolation = conn.getTransactionIsolation();
                    if (txnIsolation != Connection.TRANSACTION_READ_COMMITTED) {
                        conn.setTransactionIsolation(
                            Connection.TRANSACTION_READ_COMMITTED);
                    }
                }
            } catch ( SQLException e ) {
                gotException = true;
                throw new BrokerException(
                    br.getKString(BrokerResources.X_INTERNAL_EXCEPTION,
                    "Unable to set connection's transaction isolation level, current= "
                    + txnIsolation), e);
            }
        } finally {
            if (gotException) {
                DBConnectionPool.freeConnection(conn);
            }
        }

        return conn;
    }

    /**
     * Return the connection to the connection pool.
     */
    public void closeConnection(Connection conn) {
        if (conn != null) {
            DBConnectionPool.freeConnection(conn);
        }
    }

    /**
     * Create a new database connection either from the DataSource or
     * the DriverManager using the OPENDB_URL.
     */
    Connection newConnection(boolean autocommit) throws BrokerException {

        // make database connection; database should already exist
        Util.RetryStrategy retry = null;
        do {
            try {
                Connection conn = null;
                try {
                    if (dataSource != null) {
                        if (isPoolDataSource) {
                            if (user == null) {
                                conn = ((ConnectionPoolDataSource)dataSource)
                                    .getPooledConnection().getConnection();
                            } else {
                                conn = ((ConnectionPoolDataSource)dataSource)
                                    .getPooledConnection(user, password).getConnection();
                            }
                        } else {
                            if (user == null) {
                                conn = ((DataSource)dataSource).getConnection();
                            } else {
                                conn = ((DataSource)dataSource).getConnection(user, password);
                            }
                        }
                    } else {
                        if (user == null) {
                            conn = DriverManager.getConnection(openDBUrl);
                        } else {
                            conn = DriverManager.getConnection(openDBUrl, user, password);
                        }
                    }

                    conn.setAutoCommit(autocommit);
                } catch (Exception e) {
                    throw new BrokerException(br.getKString(
                        BrokerResources.E_CANNOT_GET_DB_CONNECTION, openDBUrl), e);
                }

                // Return the newly created connection
                return conn;
            } catch (BrokerException e) {
                // Exception will be log & re-throw if operation cannot be retry
                if ( retry == null ) {
                    retry = new Util.RetryStrategy(this);
                }
                retry.assertShouldRetry( e );
            }
        } while (true);
    }

    public boolean supportsBatchUpdates() {
        return supportBatch;
    }

    public boolean isHADB() {
        return isHADB;
    }

    public boolean isMysql() {
        return isMysql;
    }
    
    public boolean isOracle() {
        return isOracle;
    }

    public boolean isOracleDriver() {
        return isOraDriver;
    }

    public boolean isHAClusterActive(Connection conn) throws BrokerException {

        boolean isActive = false;

        BrokerDAO bkrDAO = getDBManager().getDAOFactory().getBrokerDAO();
        HashMap bkrMap = bkrDAO.getAllBrokerInfos(conn, false);
        Iterator itr = bkrMap.values().iterator();
        long currentTime = System.currentTimeMillis();
        while ( itr.hasNext() ) {
            HABrokerInfo bkrInfo = (HABrokerInfo)itr.next();
            int state = bkrInfo.getState();
            if ( !BrokerState.getState(state).isActiveState() ) {
                continue; // broker is not active
            }

            // We've a broker in active state, re-verify w/ last heartbeat;
            // If heartbeat is older than 3 minutes then consider it not active
            long lastHeartBeat = bkrInfo.getHeartbeat();
            if ( lastHeartBeat + 180000 > currentTime ) {
                isActive = true;
                break;
            }
        }

        return isActive;
    }

    public DAOFactory getDAOFactory() {
        if (daoFactory == null) {
            synchronized(lock) {
                if (daoFactory == null) {
                    // Create a DAO factory for the specified DB vendor
                    if ( isHADB ) {
                        logger.log( Logger.DEBUG, "Instantiating HADB DAO factory" );
                        daoFactory = new HADBDAOFactory();
                    } else if ( isOracle ) {
                        logger.log( Logger.DEBUG, "Instantiating Oracle DAO factory" );
                        daoFactory = new OracleDAOFactory();
                    } else {
                        logger.log( Logger.DEBUG, "Instantiating generic DAO factory" );
                        daoFactory = new GenericDAOFactory();
                    }                    
                }
            }
        }
        return daoFactory;
    }

    public String getOpenDBURL() {
        return openDBUrl;       
    }

    String getCreateDBURL() {
        return createDBUrl;
    }

    public String getBrokerID() {
        return brokerID;
    }

    String getClusterID() {
        return clusterID;
    }

    public String getUser() {
        return user;
    }

    public String getTableName(String tableNamePrefix) {
        if (tableNamePrefix == null || tableNamePrefix.length() == 0) {
            throw new NullPointerException();
        }
        return tableNamePrefix + tableSuffix;
    }

    public String getDriver() {
        return driver;
    }

    public void resetConnectionPool() throws BrokerException {
        DBConnectionPool.reset();   // Reset/clear connection pool
        DBConnectionPool.init(dbMgr);
    }
    
    void close() {
        DBConnectionPool.close();   // Clear connection pool

        synchronized(lock) {
            // Use closeDBUrl to close embedded db
            if (closeDBUrl != null) {
                try {
                    DriverManager.getConnection(closeDBUrl);
                } catch (SQLException e) {
                    if (Store.getDEBUG()) {
                        logger.log(Logger.DEBUG,
                            BrokerResources.I_DATABASE_SHUTDOWN, closeDBUrl, e);
                    }
                }
            }

            if (dbMgr != null) {
                if (dbMgr.isOraDriver) {
                    Util.OracleBLOB_initialized = false;
                }
                dbMgr = null;
            }
        }
    }

    void printInfo(PrintStream out) {
        out.println("Configuration properties:");
        out.println(driverProp + "=" + driver);
        out.println(openDBUrlProp + "=" + openDBUrl);
        out.println(createDBUrlProp + "=" + createDBUrl);
        out.println(closeDBUrlProp + "=" + closeDBUrl);
        out.println(userProp + "=" + user);
        out.println(passwordProp + "=" + password);
        out.println(needPasswordProp + "=" + needPassword);
        out.println("brokerID" + "=" + brokerID);
        out.println("clusterID" + "=" + clusterID);
    }

    // get table schema for the current store version returns a Hashtable with
    // table names map to the create table SQL
    HashMap getTableSchema() {
        return tableSchemas;
    }

    // Get all names of tables used in a specific store version; i.e. pre-4.0
    String[] getTableNames(int version) {
        String names[] = new String[0];
        if (version == JDBCStore.STORE_VERSION) {
            names = (String[]) tableSchemas.keySet().toArray(names);
        } else if (version == JDBCStore.OLD_STORE_VERSION_370) {
            names = v370tableNames;
        } else if (version == JDBCStore.OLD_STORE_VERSION_350) {
            names = v350tableNames;
        }
        return names;
    }

    // Return 0 if store has not been created, -1 if some tables are missing, or
    // a value > 0 if all tables for the store have already been created
    int checkStoreExists(Connection conn) throws BrokerException {

        Map tables = getTableNamesFromDB( conn, tableSuffix );

        int total = 0;
        int found = 0;
        Iterator itr = getDAOFactory().getAllDAOs().iterator();
        StringBuffer sbuf =  new StringBuffer();
        while ( itr.hasNext() ) {
            total ++;
            String tname = ((BaseDAO)itr.next()).getTableName();
            if ( tables.containsKey( tname.toLowerCase() ) ) {
                found++;
            } else {
                sbuf.append(tname+" ");
            }
        }

        if ( (found > 0) && (found != total) ) {
            logger.log(logger.WARNING, br.getKString(br.W_TABLE_NOT_FOUND_IN_DATABASE,
                       sbuf.toString()+"["+total+","+found+", "+tableSuffix+"]"));
            return -1;  // There is a problem! some tables are missing
        } else {
            return found;
        }
    }

    // Get all table names from the db that matchs the specified name pattern .
    // If name pattern is not specified then use tableSuffix for this broker.
    // As a precaution, only get table that starts with "mq" or "MQ".
    Map getTableNamesFromDB(Connection conn, String namePattern)
        throws BrokerException {

        // If name pattern not specified then use the table suffix
        if ( namePattern == null || namePattern.trim().length() == 0 ) {
            namePattern = tableSuffix;
        }
        namePattern = namePattern.toLowerCase();

        HashMap tableNames = new HashMap();

        Util.RetryStrategy retry = null;
        do {
            boolean myConn = false;
            ResultSet rs = null;
            try {
                // Get a connection
                if ( conn == null ) {
                    conn = getConnection( true );
                    myConn = true;
                }

                DatabaseMetaData dbMetaData = conn.getMetaData();

                String[] tTypes = { "TABLE" };
                rs = dbMetaData.getTables( null, null, null, tTypes );
                while ( rs.next() ) {
                    String tableName = rs.getString( "TABLE_NAME" );
                    if ( tableName != null ) {
                        tableName = tableName.trim();
                        if (tableName.length() == 0) { 
                            throw new BrokerException(br.getKString(
                            br.X_DB_RETURN_EMPTY_TABLENAME, "DatabaseMetaData.getTables()"));
                        }
                        if (tableName.length() < 2) { 
                            continue;
                        }
                        // Only process MQ tables
                        char ch1 = tableName.charAt(0);
                        char ch2 = tableName.charAt(1);
                        if ( ( ch1 == 'm' || ch1 == 'M' ) &&
                             ( ch2 == 'q' || ch2 == 'Q' ) ) {
                            // Saves table name that match name pattern
                            String key = tableName.toLowerCase();
                            if ( key.indexOf( namePattern ) > -1 ) {
                                tableNames.put( key, tableName );
                            }
                        }
                    }
                }

                break; // We're done so break out from retry loop
            } catch ( Exception e ) {
                // Exception will be log & re-throw if operation cannot be retry
                if ( retry == null ) {
                    retry = new Util.RetryStrategy(this);
                }
                retry.assertShouldRetry( e );
            } finally {
                if ( myConn ) {
                    Util.close( rs, null, conn );
                }
            }
        } while (true);

        return tableNames;
    }

    // Wrap SQLException with the message, usually the SQL statement
    // that generated the exception
    static SQLException wrapSQLException(String msg, SQLException e) {
        SQLException e2 = new SQLException(
            msg + ": " + e.getMessage(),
            e.getSQLState(), e.getErrorCode());
        e2.setNextException(e);
        return e2;
    }

    // Wrap IOException with the message, usually the SQL statement
    // that triggered the exception
    static IOException wrapIOException(String msg, IOException e) {
        IOException e2 = new IOException(msg + ": " + e.getMessage());
        e2.initCause(e);
        return e2;
    }

    /**
     * Lock the tables for this broker by putting our broker
     * identifier in the version table or unlock the tables by
     * setting the broker id column to null.
     * @param conn Database connection
     * @param doLock
     * @throws BrokerException if the operation is not successful because
     * it cannot be locked
     */
    static void lockTables(Connection conn, boolean doLock)
        throws BrokerException {

        // Create the lockID for this broker
        LockFile lockFile = LockFile.getCurrentLockFile();
        String lockID = null;
        if (lockFile != null) {
            lockID = lockFile.getInstance() + ":" +
                lockFile.getHost() + ":" + lockFile.getPort();
        } else {
            // we are probably imqdbmgr
            String hostName = Globals.getBrokerHostName();
            if ( hostName == null ) {
                try {
                    InetAddress ia = InetAddress.getLocalHost();
                    hostName = ia.getCanonicalHostName();
                } catch (UnknownHostException e) {
                    hostName = "";
                    Globals.getLogger().log(Logger.ERROR,
                        BrokerResources.E_NO_LOCALHOST, e);
                }
            }
            lockID = Globals.getConfigName() + ":" + hostName + ":" + "imqdbmgr";
        }

        // Check if there is a lock; if the store does not exist, VersionDAO
        // will throw a BrokerException with the status set to Status.NOT_FOUND
        VersionDAO verDAO = getDBManager().getDAOFactory().getVersionDAO();
        String currLck = verDAO.getLock( conn, JDBCStore.STORE_VERSION );
        if ( currLck == null ) {
            // We've a problem, version data not found
            if ( !doLock ) {
                Globals.getLogger().log(Logger.WARNING,
                    BrokerResources.E_BAD_STORE_NO_VERSIONDATA, verDAO.getTableName());

                // insert version info in the version table
                verDAO.insert( conn, JDBCStore.STORE_VERSION );
                return;
            } else {
                throw new BrokerException(
                    Globals.getBrokerResources().getKString(
                        BrokerResources.E_BAD_STORE_NO_VERSIONDATA,
                        verDAO.getTableName() ), Status.NOT_FOUND );
            }
        }

        boolean doUpdate = false;
        String newLockID = null;
        String oldLockID = null;
        TableLock tableLock = new TableLock( currLck );
        if ( tableLock.isNull ) {
            // NO lock!
            if ( doLock ) {
                // Try to lock the tables for this broker
                doUpdate = true;
                newLockID = lockID;
            }
        } else {
            // There is a lock
            if ( isOurLock( lockFile, tableLock ) ) {
                if ( !doLock ) {
                    // Remove the lock, i.e. setting lockID to null
                    doUpdate = true;
                    oldLockID = tableLock.lockstr;
                }
            } else {
                // Not our lock; check whether the other broker is still running
                if ( validLock( tableLock ) ) {
                    // Only action is to reset if it is lock by imqdbmgr
                    if ( !doLock && tableLock.port == 0 ) {
                        doUpdate = true;
                        oldLockID = tableLock.lockstr;
                    } else {
                        throw new BrokerException( generateLockError( tableLock ) );
                    }
                } else {
                    // Lock is invalid so take over or reset the lock
                    if ( doLock ) {
                        // Take over the lock
                        doUpdate = true;
                        newLockID = lockID;
                        oldLockID = tableLock.lockstr;
                    } else {
                        // Remove the lock, i.e. setting lockID to null
                        doUpdate = true;
                        oldLockID = tableLock.lockstr;
                    }
                }
            }
        }

        if ( doUpdate ) {
            boolean updated = verDAO.updateLock(
                conn, JDBCStore.STORE_VERSION, newLockID, oldLockID );
            if ( !updated ) {
                if ( newLockID == null ) {
                    // Failed to remove lock
                    Globals.getLogger().log( Logger.ERROR,
                        BrokerResources.E_REMOVE_STORE_LOCK_FAILED );
                }

                // Unable to get lock, e.g. another broker get to it first?
                currLck = verDAO.getLock( conn, JDBCStore.STORE_VERSION );
                throw new BrokerException(
                    generateLockError( new TableLock( currLck ) ) );
            }
        }
    }

    // check whether the string in brokerlock belong to us
    private static boolean isOurLock(LockFile lf, TableLock lock) {

        if (lf != null) {
            // i am a broker
            if (lf.getPort() == lock.port &&
                LockFile.equivalentHostNames(lf.getHost(), lock.host, false) &&
                lf.getInstance().equals(lock.instance)) {
                return true;
            } else {
                return false;
            }
        } else {
            // i am imqdbmgr; assume the lock in the table is NOT our lock
            return false;
        }
    }

    // check whether the broker specified in the lock string is
    // still running, if it is, it's a valid lock
    private static boolean validLock(TableLock lock) {

        if (lock.port == 0) {
            // locked by imqdbmgr
            return true;
        }

        Socket s = null;
        try {
            // Try opening socket to other broker's portmapper. If we
            // can open a socket then the lock is valid
            s = new Socket(lock.host, lock.port);
            return true;
        } catch (Exception e) {
            // Looks like owner is not running. the lock is not valid
            return false;
        } finally {
            if ( s != null ) {
                try {
                    s.close();
                } catch (Exception e) {}
            }
        }
    }

    private void initDataSource( Class dsClass, Object dsObject ) {

    	// Get a list of property names to initialize the Data Source,
        // e.g. imq.persist.jdbc.<dbVendor>.property.*
        String propertyPrefix = vendorPropPrefix + ".property.";
        List list = config.getPropertyNames(propertyPrefix);
        if (list.size() == 0) {
            if (Store.getDEBUG()) {
                logger.log(Logger.DEBUG, "DataSource properties not specified!");
            }
        }

        Method[] methods = dsClass.getMethods();
        Object[] arglist = new Object[1];

        // Invoke the setter method for each DataSource property
        Iterator itr = list.iterator();
        while (itr.hasNext()) {
            Exception error = null;
            String errorMsg = null;

            String propName = (String)itr.next();
            String propValue = config.getProperty(propName).trim();

            if (propValue.length() > 0) {
                String prop = propName.substring(propertyPrefix.length(),
                    propName.length());

                // Map Datasource's url property to openDBUrl
                if ( prop.equalsIgnoreCase( "url" ) ||
                     prop.equalsIgnoreCase( "serverList" ) ) {
                     openDBUrl = propValue;
                } else {
                     logger.log(Logger.FORCE, propName+"="+propValue);
                }

            	// Find the DataSource's method to set the property
                String methodName = ("set" + prop).toLowerCase();
                Method method = null; // The DataSource method
                Class paramType = null; // The param type of this method
                for (int i = 0, len = methods.length; i < len; i++) {
                    Method m = methods[i];
                    Class[] paramTypes = m.getParameterTypes();
                    if (methodName.equals(m.getName().toLowerCase()) &&
                        paramTypes.length == 1) {
                        // Found the setter method for this property
                        method = m;
                        paramType = paramTypes[0];
                        break;
                    }
                }

                if (method != null ) {
                    try {
                        if (paramType.equals( Boolean.TYPE )) {
                            arglist[0] = Boolean.valueOf(propValue);
                            method.invoke(dsObject, arglist);
                        } else if (paramType.equals(Integer.TYPE)) {
                            arglist[0] = Integer.valueOf(propValue);
                            method.invoke(dsObject, arglist);
                        } else if (paramType.equals(String.class)) {
                            arglist[0] = propValue;
                            method.invoke(dsObject, arglist);
                        } else {
                            errorMsg = "Invalid DataSource Property: " +
                            	propName + ", value: " + propValue;
                        }
                    } catch (Exception e) {
                        error = e;
                        errorMsg = "Unable to initialize DataSource Property: " +
                            propName + ", value: " + propValue;
                    }
                } else {
                    errorMsg = "Invalid DataSource Property: " + propName +
                    	", value: " + propValue;
                }
            } else {
                errorMsg = "Invalid DataSource Property: " + propName +
                	", value: " + propValue;
            }

            if (errorMsg != null) {
                logger.log(Logger.ERROR, errorMsg, error);
            }
        }
    }

    /**
     * Extract all table definition properties and put it in tables.
     * Format of the properties is:
     * - name is imq.persist.jdbc.<vendor>.table.[tablename]
     *   where [tablename] is the name of the table
     * - value is the create table SQL with the table name specified
     *   as ${name} so that we can replace it with the corresponding name
     */
    private void loadTableSchema() throws BrokerException {

        List list = config.getPropertyNames(tablePropPrefix);
        if (list.size() == 0) {
            throw new BrokerException("Table definition not found for " + vendor);
        }

        // Sort the list to ensure the table definition will come before
        // the table index definition.
        Collections.sort(list);

        // Variable ${name}, ${tableoption} and ${index} to substitute for the
        // create table SQL and create table index SQL
        Properties vars = new Properties();

        // set table option
        String key = vendorPropPrefix + "." + TABLEOPTION_NAME_VARIABLE;
        String value = config.getProperty(key, "").trim();
        vars.setProperty( TABLEOPTION_NAME_VARIABLE, value );

        if (value != null && !value.equals("")) {
            logger.log(logger.FORCE, key+"="+value);
        }

        Iterator itr = list.listIterator();
        while (itr.hasNext()) {
            key = ((String)itr.next()).trim();
            value = config.getProperty(key).trim();

            int i = key.indexOf( ".index." );
            if (i > 0) {
                // table index definition:
                //    imq.persist.jdbc.<dbVendor>.table.<name>.index.<index>=<SQL>
                int n = key.indexOf( ".table." );
                String tname = key.substring( n + 7, i ) + tableSuffix;
                String iname = tname + key.substring( i + 7 );

                // set table name & index name in the variable table so we can expand it
                vars.setProperty( TABLE_NAME_VARIABLE, tname );
                vars.setProperty( INDEX_NAME_VARIABLE, iname );
                String sql = StringUtil.expandVariables( value, vars );

                TableSchema schema = (TableSchema)tableSchemas.get( tname );
                if ( schema != null ) {
                    schema.addIndex( iname, sql );
                }
            } else {
                // table definition:
                //    imq.persist.jdbc.<dbVendor>.table.<tableName>=<SQL>
                int n = key.lastIndexOf('.');
                String tname = key.substring( n + 1 ) + tableSuffix;

                // set table name in the variable table so we can expand it
                vars.setProperty( TABLE_NAME_VARIABLE, tname );
                String sql = StringUtil.expandVariables( value, vars );

                // add table schema object
                TableSchema schema = new TableSchema( tname, sql );
                tableSchemas.put( tname, schema );
            }
        }

        checkTables();
    }

    // make sure definition for all tables needed by the broker is specified
    private void checkTables() throws BrokerException {

        Iterator itr = getDAOFactory().getAllDAOs().iterator();
        while ( itr.hasNext() ) {
            BaseDAO dao = (BaseDAO)itr.next();
            String tname = dao.getTableName();
            TableSchema schema = (TableSchema)tableSchemas.get(tname);
            if (schema == null || schema.tableSQL.length() == 0) {
                throw new BrokerException(
                    br.getKString(BrokerResources.E_NO_JDBC_TABLE_PROP, tname));
            } else {
                if (Store.getDEBUG()) {
                    logger.log(Logger.DEBUG, tname + ": '" + schema.tableSQL + "'");
                }
            }
        }
    }

    // add the brokerid to the old table names
    private void fixOldTableNames() {
        for (int i = 0; i < v370tableNames.length; i++) {
            v370tableNames[i] = v370tableNames[i] + brokerID;
        }

        for (int i = 0; i < v350tableNames.length; i++) {
            v350tableNames[i] = v350tableNames[i] + brokerID;
        }
    }

    private String getPassword() {
        // get private property first
        passwordProp = vendorPropPrefix + ".password";
        String dbpw = config.getProperty(passwordProp);
        if (dbpw == null) {
            // try fallback prop
            dbpw = config.getProperty(FALLBACK_PWD_PROP);
            if (dbpw != null) {
                passwordProp = FALLBACK_PWD_PROP;
            }
        }

        needPasswordProp = vendorPropPrefix + ".needpassword";
        if (config.getProperty(needPasswordProp) == null) {
            // try fallback prop
            String fallbackProp = JDBC_PROP_PREFIX + ".needpassword";
            if (config.getProperty(fallbackProp) != null) {
                needPasswordProp = fallbackProp;
            }
        }

        needPassword = config.getBooleanProperty(needPasswordProp, DEFAULT_NEEDPASSWORD);

        if (dbpw == null && needPassword) {
            int retry = 0;
            Password pw = new Password();
            while ((dbpw == null || dbpw.trim().equals("")) && retry < 5) {
                System.err.print(br.getString(
                    BrokerResources.M_ENTER_DB_PWD, openDBUrl));
                System.err.flush();

                dbpw = pw.getPassword();

                // Limit the number of times we try reading the passwd.
                // If the VM is run in the background the readLine()
                // will always return null and we'd get stuck in the loop
                retry++;
            }
        }

        return dbpw;
    }

    private void checkHADBJDBCLogging() {
        // Enable HADB's JDBC driver logging
        String level = config.getProperty( "com.sun.hadb.jdbc.level" );
        if (level != null && level.length() > 0) {
            String logFile = StringUtil.expandVariables(
                "${imq.instanceshome}${/}${imq.instancename}${/}log${/}hadbLog.txt",
                config);

            logger.log( Logger.INFO,
                "Enable HADB's JDBC driver logging (level=" + level + "): " +
                logFile );

            try {
                // Setup logger
                java.util.logging.Logger jLogger =
                    java.util.logging.Logger.getLogger( "com.sun.hadb.jdbc" );
                java.util.logging.FileHandler fh =
                    new java.util.logging.FileHandler(logFile, true);
                fh.setFormatter(new java.util.logging.SimpleFormatter());
                jLogger.addHandler(fh);
                jLogger.setLevel(java.util.logging.Level.parse(level));
            } catch (Exception e) {
                logger.logStack( Logger.WARNING,
                    "Failed to enable HADB's JDBC driver logging", e );
            }
        }
    }

    private static String generateLockError(TableLock lock) {
        // locked by jmqdbmgr
        BrokerResources br = Globals.getBrokerResources();
        if (lock.port != 0) {
            return br.getKString(BrokerResources.E_TABLE_LOCKED_BY_BROKER,
                                lock.host, String.valueOf(lock.port));
        } else {
            return br.getKString(BrokerResources.E_TABLE_LOCKED_BY_DBMGR);
        }
    }

    private static class TableLock {
        String lockstr;
        String instance;
        String host;
        String portstr;
        int port = 0;
        boolean isNull = false;

        TableLock(String str) {
            lockstr = str;

            if (lockstr == null || lockstr.length() == 0) {
                isNull = true;
            } else {
                // lockstr has the format <instance>:<host>:<port>
                StringTokenizer st = new StringTokenizer(lockstr, " :\t\n\r\f");

                try {
                    instance = st.nextToken();
                    host = st.nextToken();
                    portstr = st.nextToken();
                    try {
                        port = Integer.parseInt(portstr);
                    } catch (NumberFormatException e) {
                        port = 0;
                    }
                } catch (NoSuchElementException e) {
                    Globals.getLogger().log(Logger.WARNING, 
                        BrokerResources.W_BAD_BROKER_LOCK, lockstr);
                    lockstr = null;
                    isNull = true;
                }
            }
        }
    }

    static final class TableSchema {
        String tableName;
        String tableSQL;
        HashMap indexMap = new HashMap();

        TableSchema(String name, String sql) {
            tableName = name;
            tableSQL = sql;
        }

        void addIndex(String name, String sql) {
            if (sql != null && sql.trim().length() > 0) {
                indexMap.put(name, sql);
            } else {
                Globals.getLogger().log( Logger.WARNING,
                    "SQL command to create the table index " + name +
                    " for table " + tableName + " is null or empty" );
            }
        }

        String getIndex(String name) {
            return (String)indexMap.get(name);
        }

        Iterator indexIterator() {
            return indexMap.keySet().iterator();
        }
    }
}
