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

/*
 * @(#)DBTool.java	1.108 06/29/07
 */ 

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

import com.sun.messaging.jmq.Version;
import com.sun.messaging.jmq.io.Status;
import com.sun.messaging.jmq.jmsserver.config.*;
import com.sun.messaging.jmq.jmsserver.util.*;
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.util.log.Logger;
import com.sun.messaging.jmq.util.StringUtil;
import com.sun.messaging.jmq.util.UID;

import com.sun.messaging.jmq.jmsserver.persist.StoreManager;
import com.sun.messaging.jmq.jmsserver.persist.HABrokerInfo;
import com.sun.messaging.jmq.jmsserver.persist.TransactionInfo;
import com.sun.messaging.jmq.jmsserver.persist.file.FileStore;
import com.sun.messaging.jmq.jmsserver.core.Destination;
import com.sun.messaging.jmq.jmsserver.core.DestinationUID;
import com.sun.messaging.jmq.jmsserver.core.Consumer;
import com.sun.messaging.jmq.jmsserver.core.ConsumerUID;
import com.sun.messaging.jmq.jmsserver.data.TransactionUID;
import com.sun.messaging.jmq.jmsserver.data.TransactionAcknowledgement;
import com.sun.messaging.jmq.io.Packet;
import com.sun.messaging.jmq.io.SysMessageID;
import com.sun.messaging.bridge.BridgeServiceManager;
import com.sun.messaging.bridge.service.JMSBridgeStore;

import java.io.*;
import java.sql.*;
import java.util.*;

/**
 * This class is used to create, delete, and recreate database tables.
 * It may also be used to create a database.
 * All database specific information is obtained from property file,
 * except for username and password which may also be specified as command
 * line arguments.
 */
public class DBTool implements DBConstants {

    static final String SQLFILEDIR_PROP =
		DBManager.JDBC_PROP_PREFIX + "sqlfile.dirpath";

    static final String SQLFILENAME_PROP =
		DBManager.JDBC_PROP_PREFIX + "sqlfile.name";

    /*
     * All internal properties must start with "imq."
     */
    static final String STORE_PROPERTY_HABROKERS = "imq.cluster.haBrokers";
    static final String STORE_PROPERTY_SUPPORT_JMSBRIDGE = "imq.bridge.jmsbridge.tables";

    // parser exception reasons
    private static int EXTRA_CMD_SPECIFIED	= 0;
    private static int BAD_CMD_ARG		= 1;
    private static int BAD_OPT			= 2;
    private static int BAD_OPT_ARG		= 3;
    private static int MISSING_OPT_ARG		= 4;
    private static int MISSING_CMD_ARG		= 5;
    private static int MISSING_OPT          = 6;

    private static String CMD_NAME		= "dbmgrcmd";
    private static String CREATE_ALL_CMD	= "createall";
    private static String CREATE_TBL_CMD	= "createtbl";
    private static String DELETE_TBL_CMD	= "deletetbl";
    private static String RECREATE_TBL_CMD	= "recreatetbl";
    private static String REMOVE_BKR_CMD	= "removebkr";
    private static String REMOVE_JMSBRIDGE_CMD = "removejmsbridge";
    private static String DUMP_CMD		= "dump";
    private static String DROPTBL_CMD	        = "droptbl";
    private static String RESET_CMD		= "reset";
    private static String BACKUP_CMD            = "backup";
    private static String RESTORE_CMD           = "restore";
    private static String UPGRADE_STORE_CMD     = "upgradestore";
    private static String UPGRADE_HASTORE_CMD   = "upgradehastore";
    private static String QUERY_CMD             = "query";

    private static String ARG_NAME		= "dbmgrarg";

    private static String CREATE_CMD_STR = "create";
    private static String DELETE_CMD_STR = "delete";
    private static String RECREATE_CMD_STR = "recreate";
    private static String REMOVE_CMD_STR = "remove";
    private static String UPGRADE_CMD_STR = "upgrade";
    private static String ARGU_ALL = "all";
    private static String ARGU_TBL = "tbl";
    private static String ARGU_BKR = "bkr";
    private static String ARGU_JMSBRIDGE = "jmsbridge";
    private static String ARGU_OLDTBL = "oldtbl";
    private static String ARGU_LCK = "lck";
    private static String ARGU_STORE = "store";
    private static String ARGU_HASTORE = "hastore";
    private static String OPT_H = "-h";
    private static String OPT_LH = "-help";
    private static String OPT_V = "-v";
    private static String OPT_LV = "-version";
    private static String OPT_B = "-b";
    private static String OPT_N = "-n";
    private static String OPT_U = "-u";
    private static String OPT_P = "-p";
    private static String OPT_PW = "-pw";
    private static String OPT_PASSFILE = "-passfile";
    private static String OPT_D = "-D";
    private static String OPT_VARHOME = "-varhome";
    private static String OPT_VERBOSE = "-verbose";
    private static String OPT_DEBUG = "-debug";
    private static String OPT_DIR = "-dir";
    private static String OPT_FORCE = "-f";
    private static String JMSBRIDGE_NAME_PROPERTY = "jmsbridge.name";

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

    private Version version;
    private DBManager dbmgr = null;
    private boolean standalone = true;
    private boolean cliPasswdSpecified = false;
    private boolean debugSpecified = false;
    private boolean forceSpecified = false;

    DBTool(boolean standalone) {
	this.version = new Version();
	this.standalone = standalone;
    }

    private void doCreate(boolean createdb)
	throws BrokerException {

	Connection conn = null;
	try {
            if (createdb) {
                conn = dbmgr.connectToCreate();
                conn.setAutoCommit( true );
            } else {
                conn = dbmgr.newConnection( true ); // set autoCommit to true
            }

            // Check if store exist
            boolean continueOnError = false;
            int status = dbmgr.checkStoreExists( conn );
            if (status > 0) {
                // All tables have already been created
                throw new BrokerException(br.getKString(
                    BrokerResources.E_DATABASE_TABLE_ALREADY_CREATED));
            } else if (status < 0) {
                // Some tables are missings so try to create the tables
                // but ignore error if table already exists
                continueOnError = true;
            }

            createTables( conn, continueOnError );

	    if (standalone) {
                if ( Globals.getHAEnabled() ) {
                    System.out.println( br.getString(
                        BrokerResources.I_DATABASE_TABLE_HA_CREATED,
                        Globals.getClusterID() ) );
                } else {
		    System.out.println(
                        br.getString(BrokerResources.I_DATABASE_TABLE_CREATED));
                }
	    }
	} catch (Throwable t) {
	    String url;
	    if (createdb)
		url = dbmgr.getCreateDBURL();
	    else
		url = dbmgr.getOpenDBURL();

	    throw new BrokerException(br.getKString(
                BrokerResources.E_CREATE_DATABASE_TABLE_FAILED, url, t));
	} finally {
            if ( !createdb && conn != null ) {
                // Since the connection is not from the pool; we've to close it
                try {
                    conn.close();
                } catch (SQLException e) {
                    throw new BrokerException(
                        Globals.getBrokerResources().getKString(
                            BrokerResources.E_INTERNAL_BROKER_ERROR,
                            "Unable to close JDBC resources", e ) );
                }
            }
	}
    }

    // create database tables used in the current version of persistent store
    static void createTables( Connection conn ) throws BrokerException {
        createTables( conn, false );
    }

    static void createTables( Connection conn, boolean continueOnError )
        throws BrokerException {
        createTables(conn, continueOnError, null);
    }

    //if tableDAOs != null, only create these tables 
    //else create current version of persist store
    static void createTables( Connection conn, boolean continueOnError, 
                              ArrayList tableDAOs)
        throws BrokerException {

        DBManager dbMgr = DBManager.getDBManager();
        DAOFactory daoFactory = dbMgr.getDAOFactory();
        Iterator itr = null; 
        if (tableDAOs != null) {
            itr = tableDAOs.iterator();
        } else {
            itr = daoFactory.getAllDAOs().iterator();
        }
        while ( itr.hasNext() ) {
            BaseDAO dao = (BaseDAO)itr.next();
            try {
                Util.RetryStrategy retry = null;
                do {
                    try {
                        dao.createTable( conn );
                        break; // table created so break from retry loop
                    } catch ( Exception e ) {
                        // Exception will be log & re-throw if operation cannot be retry
                        if ( retry == null ) {
                            retry = new Util.RetryStrategy();
                        }
                        retry.assertShouldRetry( e );
                    }
                } while ( true );
            } catch (BrokerException be) {
                if (Globals.getHAEnabled()) {
                    // Paused for a few secs to prevent race condition when two
                    // or more brokers try to create the tables at the same time
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {}
                }

                // Verify if the store has already been created
                if ( dbMgr.checkStoreExists( conn ) > 0 ) {
                    if (tableDAOs == null) {
                    Globals.getLogger().log(Logger.WARNING,
                        BrokerResources.E_CREATE_DATABASE_TABLE_FAILED,
                        Globals.getBrokerResources().getString(
                            BrokerResources.E_DATABASE_TABLE_ALREADY_CREATED));
                    continueOnError = true;
                    } else {
                    Globals.getLogger().log(Logger.WARNING,
                        BrokerResources.E_CREATE_DATABASE_TABLE_FAILED,
                        Globals.getBrokerResources().getString(
                            BrokerResources.E_THE_DATABASE_TABLE_ALREADY_CREATED, dao.getTableName()));
                    }
                    break;
                } else if (continueOnError) {
                    // Just log msg and continue
                    Globals.getLogger().log(Logger.WARNING, be.toString(), be.getCause());
                } else {
                    throw be;
                }
            }
        }

        if (tableDAOs != null) return;

	// Insert version info in the version table
        VersionDAO versionDAO = daoFactory.getVersionDAO();
        try {
            if ( continueOnError ) {
                // Do this only if version is missing from version table
                int storeVersion = versionDAO.getStoreVersion( conn );
                if ( storeVersion != JDBCStore.STORE_VERSION ) {
                    versionDAO.insert( conn, JDBCStore.STORE_VERSION );
                }
            } else {
                versionDAO.insert( conn, JDBCStore.STORE_VERSION );
            }
        } catch (BrokerException be) {
            if ( Globals.getHAEnabled() ) {
                // Re-check if version info has been added by another broker
                int storeVersion = versionDAO.getStoreVersion( conn );
                if ( storeVersion != JDBCStore.STORE_VERSION ) {
                    throw be; // Re-throw the exception
                }
            } else {
                throw be; // Re-throw the exception
            }
        }

        // Insert a unique store session for the stand-alone broker
        if ( !Globals.getHAEnabled() ) {
            // Do this only if store session is missing from session table
            String brokerID = Globals.getBrokerID();
            StoreSessionDAO sessionDAO = daoFactory.getStoreSessionDAO();
            if ( sessionDAO.getStoreSession( conn, brokerID ) <= 0 ) {
                sessionDAO.insert( conn, brokerID, new UID().longValue() );
            }
        }

        PropertyDAO dao = daoFactory.getPropertyDAO();
        dao.update( conn, STORE_PROPERTY_SUPPORT_JMSBRIDGE, Boolean.valueOf(true)); 
    }

    static void updateStoreVersion410IfNecessary(Connection conn)
        throws BrokerException {

        try {
            if (!Globals.getHAEnabled()) {
               DBManager.lockTables(conn, true);
            }
            DBManager dbMgr = DBManager.getDBManager();
            DAOFactory daoFactory = dbMgr.getDAOFactory();
            VersionDAO versionDAO  = daoFactory.getVersionDAO();
            int storeVersion = versionDAO.getStoreVersion(conn);

            if (storeVersion != JDBCStore.STORE_VERSION) {
                throw new BrokerException(br.getKString(
                          BrokerResources.E_BAD_STORE_VERSION,
                          String.valueOf(storeVersion), 
                          String.valueOf(JDBCStore.STORE_VERSION)));
            }
            PropertyDAO dao = daoFactory.getPropertyDAO();
            if (!dao.hasProperty(conn, STORE_PROPERTY_SUPPORT_JMSBRIDGE)) {
                ArrayList daos = new ArrayList();
                daos.add(daoFactory.getTMLogRecordDAOJMSBG());
                daos.add(daoFactory. getJMSBGDAO());
                createTables(conn, false, daos);

                dao.update( conn, STORE_PROPERTY_SUPPORT_JMSBRIDGE, Boolean.valueOf(true)); 
            }

        } finally {
            if (!Globals.getHAEnabled()) {
                DBManager.lockTables(conn, false);
            }
        }
    }

    private void doReset() throws BrokerException {

	Connection conn = null;
	try {
            conn = dbmgr.getConnection(true);

            // unlock the tables
	    DBManager.lockTables(conn, false); // false = unlock
	} finally {
            Util.close( null, null, conn );
	}
    }

    private void doDelete(String arg) throws BrokerException {

	boolean deleted = false;
        Connection conn = null;
	try {
            conn = dbmgr.getConnection(true);

	    if (arg == null || arg.length() == 0) {
                // Check if store exist
                boolean continueOnError = false;
                int status = dbmgr.checkStoreExists( conn );
                if (status > 0) {
                    // Verify cluster is not active in HA mode
                    if (!forceSpecified && Globals.getHAEnabled() &&
                        dbmgr.isHAClusterActive(conn)) {
                        throw new BrokerException(br.getKString(
                            BrokerResources.E_HA_CLUSTER_STILL_ACTIVE, dbmgr.getClusterID()));
                    }

                    try {
                        // lock the tables first (implemented since version 350);
                        // note that we don't need to unlock the tables since
                        // the tables will be dropped when we are done
                        if (!forceSpecified) {
                            DBManager.lockTables(conn, true); // true = lock
                        }
                    } catch ( BrokerException e ) {
                        if ( e.getStatusCode() == Status.NOT_FOUND ) {
                            // For some reason if version table doesn't exist or
                            // version data is not found we can just ignore the error!
                            continueOnError = true;
                        } else {
                            throw e;
                        }
                    }
                } else if (status < 0) {
                    // Some tables are missings so try to delete the rest
                    // but ignore error if table does not exists
                    continueOnError = true;
                } else {
                    // All tables have already been deleted
                    throw new BrokerException(br.getKString(
                        BrokerResources.E_DATABASE_TABLE_ALREADY_DELETED));
                }

		deleted = dropTables(conn, null, continueOnError);
	    } else if (arg.equals(ARGU_OLDTBL)) {
                int oldStoreVersion = -1;
                if ( checkVersion( conn,
                    VERSION_TBL_37 + dbmgr.getBrokerID() ) ) {
                    oldStoreVersion = JDBCStore.OLD_STORE_VERSION_370;
                } else if ( checkVersion( conn,
                    VERSION_TBL_35 + dbmgr.getBrokerID() ) ) {
                    oldStoreVersion = JDBCStore.OLD_STORE_VERSION_350;
                } else {
                    throw new BrokerException("Old persistent store (version " +
                        JDBCStore.OLD_STORE_VERSION_370 + ") not found");
                }

                deleted = dropTables(conn, dbmgr.getTableNames(oldStoreVersion));
	    } else {
		// not possible since argument is checked already
	    }

	    if (standalone && deleted) {
                if ( Globals.getHAEnabled() ) {
                    System.out.println( br.getString(
                        BrokerResources.I_DATABASE_TABLE_HA_DELETED,
                        Globals.getClusterID() ) );
                } else {
		    System.out.println(br.getString(BrokerResources.I_DATABASE_TABLE_DELETED));
                }
	    }
	} catch (Exception e) {
	    throw new BrokerException(
                br.getKString(BrokerResources.E_DELETE_DATABASE_TABLE_FAILED,
                    dbmgr.getOpenDBURL()), e);
	} finally {
            Util.close( null, null, conn );
	}
    }

    static boolean dropTables(Connection conn, String names[])
        throws SQLException, BrokerException {
        return dropTables(conn, names, false);
    }

    static boolean dropTables(Connection conn, String names[], boolean continueOnError)
        throws SQLException, BrokerException {

        boolean deleted = false;
        try {
            if ( names != null && names.length > 0 ) {
                // Removing old tables, i.e. 3.5
                Statement stmt = null;
                try {
                    stmt = conn.createStatement();
                    for ( int i = 0, len = names.length; i < len; i++ ) {
                        String oldTableName = names[i];
                        Globals.getLogger().logToAll( Logger.INFO,
                            br.getString( BrokerResources.I_DROP_TABLE,
                                oldTableName ) );
                        stmt.executeUpdate( "DROP TABLE " + oldTableName );
                    }
                } finally {
                    Util.close( null, stmt, null );
                }
            } else {
                DAOFactory daoFactory = DBManager.getDBManager().getDAOFactory();
                Iterator itr = daoFactory.getAllDAOs().iterator();
                while ( itr.hasNext() ) {
                    BaseDAO dao = (BaseDAO)itr.next();
                    try {
                        Util.RetryStrategy retry = null;
                        do {
                            try {
                                dao.dropTable( conn );
                                break; // table dropped so break from retry loop
                            } catch ( Exception e ) {
                                // Exception will be log & re-throw if operation cannot be retry
                                if ( retry == null ) {
                                    retry = new Util.RetryStrategy();
                                }
                                retry.assertShouldRetry( e );
                            }
                        } while ( true );
                    } catch (BrokerException be) {
                        if (continueOnError) {
                            // Just log msg and continue
                            Globals.getLogger().log(Logger.WARNING, be.toString(), be.getCause());
                        } else {
                            throw be;
                        }
                    }
                }
            }
            deleted = true;
        } catch ( SQLException e ) {
            Globals.getLogger().log(Logger.ERROR,
                BrokerResources.E_DELETE_DATABASE_TABLE_FAILED,
                e.toString(), e);
            throw e;
        } catch ( BrokerException e ) {
            Globals.getLogger().log(Logger.ERROR,
                BrokerResources.E_DELETE_DATABASE_TABLE_FAILED,
                e.toString(), e);
            throw e;
        }

	return deleted;
    }

    static boolean checkVersion(Connection conn, String vtable)
        throws BrokerException {

        // get version from table
        String selectSQL = "SELECT * FROM " + vtable;

        Statement stmt = null;
        ResultSet rs = null;
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery(selectSQL);
        } catch (SQLException e) {
            // assume that the table does not exist
            return false;
        } finally {
            Util.close(rs, stmt, null);
        }

        return true;
    }

    private void doRecreate() throws BrokerException {

	Connection conn = null;
	try {
            conn = dbmgr.getConnection(true);

            boolean deleted = false;
            boolean continueOnError = false;
            int status = dbmgr.checkStoreExists( conn );
            if (status > 0) {
                // Verify cluster is not active in HA mode
                if (!forceSpecified && Globals.getHAEnabled() &&
                    dbmgr.isHAClusterActive(conn)) {
                    throw new BrokerException(br.getKString(
                        BrokerResources.E_HA_CLUSTER_STILL_ACTIVE, dbmgr.getClusterID()));
                }

                try {
                    // lock the tables first (implemented since version 350);
                    // note that we don't need to unlock the tables since
                    // the tables will be recreated when we are done
                    if (!forceSpecified) {
                        DBManager.lockTables(conn, true); // true = lock
                    }
                } catch (BrokerException e) {
                    if ( e.getStatusCode() == Status.NOT_FOUND ) {
                        // For some reason if version table doesn't exist or
                        // version data is not found we can just ignore the error!
                        continueOnError = true;
                    } else {
                        // unable to get the lock
                        throw e;
                    }
                }
            } else if (status < 0) {
                // Some tables are missings so try to delete the rest
                // but ignore error if table does not exists
                continueOnError = true;
            } else {
                deleted = true;
            }

            if ( !deleted ) {
                dropTables( conn, null, continueOnError );
            }

	    createTables( conn, false );

	    if (standalone) {
                if ( Globals.getHAEnabled() ) {
                    System.out.println( br.getString(
                        BrokerResources.I_DATABASE_TABLE_HA_CREATED,
                        Globals.getClusterID() ) );
                } else {
		    System.out.println(
                        br.getString(BrokerResources.I_DATABASE_TABLE_CREATED) );
                }
	    }
	} catch (Throwable t) {
	    throw new BrokerException(
                br.getKString(BrokerResources.E_RECREATE_DATABASE_TABLE_FAILED,
                    dbmgr.getOpenDBURL()), t);
	} finally {
            Util.close( null, null, conn );
	}
    }

    private void doRemoveBkr() throws BrokerException {

        String brokerID = Globals.getBrokerID();
	Connection conn = null;
        String errorMsg = null;
	try {
            conn = dbmgr.getConnection(true);

            // try to lock the tables first
            DBManager.lockTables(conn, true); // true = lock

            try {
                DAOFactory daoFactory = dbmgr.getDAOFactory();
                BrokerDAO brokerDAO = daoFactory.getBrokerDAO();
                BrokerState state = brokerDAO.getState(conn, brokerID);
                if ( !state.isActiveState() ) {
                    try {
                        // Remove broker's and its data
                        conn.setAutoCommit(false); // do this in 1 txn
                        daoFactory.getConsumerStateDAO().deleteAll(conn);
                        daoFactory.getMessageDAO().deleteAll(conn);
                        daoFactory.getDestinationDAO().deleteAll(conn);
                        daoFactory.getTransactionDAO().deleteAll(conn);
                        daoFactory.getStoreSessionDAO().deleteByBrokerID(conn, brokerID);
                        brokerDAO.delete(conn, brokerID);
                        daoFactory.getTMLogRecordDAOJMSBG().deleteAll(conn);
                        daoFactory.getJMSBGDAO().deleteAll(conn);
                        conn.commit(); // commit changes
                    } finally {
                        conn.setAutoCommit(true);
                    }
                } else {
                    errorMsg = br.getString(
                        BrokerResources.E_REMOVE_BROKER_FAILED,
                        brokerID, "broker is still active - " + state );
                }
            } finally {
                // remove the lock
                DBManager.lockTables(conn, false); // false = unlock
            }
	} catch (Exception e) {
	    throw new BrokerException(
		br.getKString( BrokerResources.E_REMOVE_BROKER_2_FAILED,
                    brokerID ), e );
	} finally {
            Util.close( null, null, conn );
	}

        if ( standalone ) {
            if ( errorMsg == null ) {
                System.out.println(
                    br.getString( BrokerResources.I_BROKER_REMOVE, brokerID ) );
            } else {
                System.out.println(errorMsg);
            }
        }
    }

    private void doRemoveJMSBridge(String bname) throws BrokerException {

        if (bname == null) {
            try {
            checkArg(null, null, OPT_N, 0, 0);
            } catch (ParserException e) {
            handleParserException(e);
            exit(1);
            }
        }

        String brokerID = Globals.getBrokerID();
        Connection conn = null;
        String errorMsg = null;
        try {

            if (!askConfirmation("Removing JMS Bridge "+bname+".")) {
                return; 
            }

            conn = dbmgr.getConnection(true);

            DBManager.lockTables(conn, true); // true = lock

            try {
                DAOFactory daoFactory = dbmgr.getDAOFactory();
                BrokerDAO brokerDAO = daoFactory.getBrokerDAO();
                BrokerState state = brokerDAO.getState(conn, brokerID);
                if ( !state.isActiveState() ) {
                    try {
                        conn.setAutoCommit(false); 
                        daoFactory.getJMSBGDAO().delete(conn, bname, null);
                        daoFactory.getTMLogRecordDAOJMSBG().deleteAllByName(conn, bname, null);
                        conn.commit();
                    } finally {
                        conn.setAutoCommit(true);
                    }
                } else {
                    errorMsg = br.getString(BrokerResources.E_BROKER_STILL_ACTIVE);
                }
            } finally {
                DBManager.lockTables(conn, false);
            }
	    } catch (Exception e) {
            throw new BrokerException(br.getString(
            br.E_REMOVE_JMSBRIDGE_FAILED, bname, e.getMessage()), e);
	    } finally {
            Util.close( null, null, conn );
	    }

        if ( standalone ) {
            if ( errorMsg == null ) {
                System.out.println(br.getString(
                BrokerResources.I_JMSBRIDGE_REMOVED_FROM_HA_STORE, bname ));
            } else {
                System.out.println(errorMsg);
            }
        }
    }

    private void doUpgrade(boolean haStore)
	throws BrokerException {

	Connection conn = null;
	try {
            conn = dbmgr.getConnection(true);

            if ( haStore ) {
                new UpgradeHAStore().upgradeStore( conn );
            } else {
                JDBCStore store = (JDBCStore)StoreManager.getStore();
                int oldStoreVersion = -1;
                if ( store.checkOldStoreVersion( conn,
                    dbmgr.getTableName(VersionDAO.TABLE + DBConstants.SCHEMA_VERSION_40),
                    VersionDAO.STORE_VERSION_COLUMN,
                    JDBCStore.OLD_STORE_VERSION_400 ) ) {
                    oldStoreVersion = JDBCStore.OLD_STORE_VERSION_400;
                } else if ( store.checkOldStoreVersion( conn,
                    VERSION_TBL_37 + dbmgr.getBrokerID(), TVERSION_CVERSION,
                    JDBCStore.OLD_STORE_VERSION_370 ) ) {
                    oldStoreVersion = JDBCStore.OLD_STORE_VERSION_370;
                } else if ( store.checkOldStoreVersion( conn,
                    VERSION_TBL_35 + dbmgr.getBrokerID(), TVERSION_CVERSION,
                    JDBCStore.OLD_STORE_VERSION_350 ) ) {
                    oldStoreVersion = JDBCStore.OLD_STORE_VERSION_350;
                } else {
                    throw new BrokerException("Old persistent store (version " +
                        JDBCStore.OLD_STORE_VERSION_400 + ") not found");
                }

                new UpgradeStore(store, oldStoreVersion).upgradeStore(conn);
            }
	} finally {
            Util.close( null, null, conn );
	}
    }

    /**
     * dump info about the database.
     * for debugging purpose
     */
    private void doDump(String arg) throws SQLException, BrokerException {
	int storeVersion = JDBCStore.STORE_VERSION;

	if (arg != null) {
	    try {
		storeVersion = Integer.parseInt(arg);
	    } catch (NumberFormatException e) {
		storeVersion = 0;
	    }
	}

        String names[] = null;
        if ( storeVersion != JDBCStore.STORE_VERSION ) {
            // Old store
            names = dbmgr.getTableNames(storeVersion);
            if (names == null || names.length == 0) {
                throw new BrokerException("version " + arg + " not supported");
            }
        }

        Connection conn = null;
        try {
            conn = dbmgr.getConnection(true);
            doDump(conn, names);
        } finally {
            Util.close( null, null, conn );
        }
    }

    static void doDump(Connection dbconn, String tables[])
	throws BrokerException, SQLException {

        if ( tables != null && tables.length > 0 ) {
	    Statement stmt = null;
            try {
                stmt = dbconn.createStatement();
                for (int i = 0; i < tables.length; i++) {
                    ResultSet r = null;
                    String tname = tables[i];
                    String sql = "SELECT COUNT(*) FROM " + tname;
                    try {
                        r = stmt.executeQuery(sql);
                        if (r.next()) {
                            System.out.println(tname + ": number of row="
                                                + r.getInt(1));
                        }
                        r.close();
                    } catch (SQLException e) {
                        SQLException ex = DBManager.wrapSQLException(
                                                        "[" + sql + "]", e);
                        logger.log(Logger.ERROR, "failed to dump tables", ex);
                    }
                }
            } finally {
                Util.close( null, stmt, null );
            }
        } else {
            // Starting in 4.0, use info from BaseDAO.getDebugInfo()
            DAOFactory daoFactory = DBManager.getDBManager().getDAOFactory();
            Iterator itr = daoFactory.getAllDAOs().iterator();
            while ( itr.hasNext() ) {
                BaseDAO dao = (BaseDAO)itr.next();
                System.out.println( dao.getDebugInfo( dbconn ).toString() );

            }
        }
    }

    /**
     * drop tables that start with mq and match the specified table name pattern.
     * for internal use!!!
     */
    private void doDropTablesByPattern(String arg) throws BrokerException {

        Connection conn = null;
        try {
            conn = dbmgr.getConnection(true);
            Map map = dbmgr.getTableNamesFromDB(conn, arg);
            String[] names = (String[])map.values().toArray(new String[0] );

            if ( names == null || names.length == 0 ) {
                System.out.println(
                    "There were no tables that match the name pattern 'MQ*" +
                    arg + "'.");
                return;
            }

            // List all tables that match the name pattern
            System.out.println(
                "Tables matching the name pattern 'MQ*" + arg + "':");

            for (int i = 0, len = names.length; i < len; i++) {
                System.out.println("\t" + names[i]);
            }

            if (!askConfirmation("Remove all tables that match 'MQ*"+ arg +"'.")) {
                // if not positive confirmation, do nothing!
                return;
            }

            dropTables(conn, names, true);
        } catch (Exception e) {
            System.err.println(
                "Failed to drop tables by name pattern: " + e.getMessage());
            throw new BrokerException(
                br.getKString(BrokerResources.E_DELETE_DATABASE_TABLE_FAILED,
                    "table name pattern '" + arg + "'"), e);
        } finally {
            Util.close( null, null, conn );
        }
    }


    private boolean askConfirmation(String prompt) throws IOException {
        // Ask for confirmation only if force option is not specified
        if (forceSpecified) return true;

        System.out.println(prompt+"\nDo you wish to proceed? [y/n] ");
        System.out.flush();

        String val = (new BufferedReader(new InputStreamReader(System.in))).readLine();

        if (!"y".equalsIgnoreCase(val) && !"yes".equalsIgnoreCase(val)) {
            return false;
        }
        return true;
    }


    /**
     * Backup the JDBC store to filebased backup files.
     */
    private void doBackup() throws BrokerException {

        if (!Globals.getHAEnabled()) {
            throw new BrokerException(br.getKString(BrokerResources.I_HA_NOT_ENABLE,
                dbmgr.getBrokerID()));
        }

        // instantiate the file store.
        Properties props = System.getProperties();
        String clusterID = Globals.getClusterID();
        String backupDir = (String)props.get(Globals.IMQ + ".backupdir") +
            File.separator + clusterID;

        logger.logToAll(Logger.INFO,
            "Backup persistent store for HA cluster " + clusterID);

        FileStore fileStore = new FileStore(backupDir, false);

        // for backup, need to clear the store before storing anything.
        fileStore.clearAll(false);

        JDBCStore jdbcStore = null;
        try {
            jdbcStore = (JDBCStore)StoreManager.getStore();

            /*
             * For data that are not broker specific, i.e. properties, change
             * records, consumers, brokers, and sessions, we will store
             * those data on the top level directory (a.k.a cluster filestore).
             */

            // Properties table
            Properties properties = jdbcStore.getAllProperties();
            Iterator propItr = properties.entrySet().iterator();
            while (propItr.hasNext()) {
                Map.Entry entry = (Map.Entry)propItr.next();
                fileStore.updateProperty(
                    (String)entry.getKey(), entry.getValue(), false);

            }
            propItr = null; properties = null;

            // Configuration Change Record table
            Object[] recordArray = jdbcStore.getAllConfigRecords();
            ArrayList tsList = (ArrayList)recordArray[0];
            ArrayList recList = (ArrayList)recordArray[1];
            for (int i = 0, len = tsList.size(); i < len; i++) {
                fileStore.storeConfigChangeRecord(
                    ((Long)tsList.get(i)).longValue(),
                    (byte[])recList.get(i), false);
            }
            recordArray = null; tsList = null; recList = null;

            // Consumer table
            Consumer[] consumerArray = jdbcStore.getAllInterests();
            for (int i = 0; i < consumerArray.length; i++) {
                fileStore.storeInterest(consumerArray[i], false);
            }
            consumerArray = null;

            // Broker & Session table - store all HABrokerInfo as a property
            HashMap bkrMap = jdbcStore.getAllBrokerInfos(true);
            List haBrokers = new ArrayList(bkrMap.values());
            fileStore.updateProperty(STORE_PROPERTY_HABROKERS, haBrokers, true);

            /*
             * For each broker in the cluster, we will store broker specific
             * data, destinations, messages, transactions, acknowledgements
             * in their own filestore (a.k.a broker filestore).
             */
            Iterator bkrItr = haBrokers.iterator();
            while (bkrItr.hasNext()) {
                // Backup data for each broker
                HABrokerInfo bkrInfo = (HABrokerInfo)bkrItr.next();
                String brokerID = bkrInfo.getId();

                logger.logToAll(Logger.INFO,
                    "Backup persistent data for broker " + brokerID);

                FileStore bkrFS = null;
                JMSBridgeStore jmsbridgeStore = null;
                try {
                    String instanceRootDir = backupDir + File.separator + brokerID;
                    bkrFS = new FileStore(instanceRootDir, false);
                    bkrFS.clearAll(false);

                    // Destination table.
                    Destination[] dstArray = jdbcStore.getAllDestinations(brokerID);
                    for (int i = 0, len = dstArray.length; i < len; i++) {
                        DestinationUID did = dstArray[i].getDestinationUID();
                        Destination dst = bkrFS.getDestination(did);
                        if (dst == null) {
                            // Store the destination if not found
                            bkrFS.storeDestination(dstArray[i],  false);
                        }
                    }

                    // Storing messages for each destination.
                    for (int i = 0, len = dstArray.length; i < len; i++) {
                        for (Enumeration e = jdbcStore.messageEnumeration(dstArray[i]);
                            e.hasMoreElements();) {
                            DestinationUID did = dstArray[i].getDestinationUID();
                            Packet message = (Packet)e.nextElement();
                            SysMessageID mid = message.getSysMessageID();

                            // Get interest states for the message; Consumer State table
                            HashMap stateMap = jdbcStore.getInterestStates(did, mid);
                            if (stateMap == null || stateMap.isEmpty()) {
                                bkrFS.storeMessage(did, message, false);
                            } else {
                                int size = stateMap.size();
                                ConsumerUID[] iids = new ConsumerUID[size];
                                int[] states  = new int[size];
                                Iterator stateItr = stateMap.entrySet().iterator();
                                int j = 0;
                                while (stateItr.hasNext()) {
                                    Map.Entry entry = (Map.Entry)stateItr.next();
                                    iids[j] = (ConsumerUID)entry.getKey();
                                    states[j] = ((Integer)entry.getValue()).intValue();
                                    j++;
                                }
                                bkrFS.storeMessage(did, message, iids, states, false);
                            }
                        }
                    }

                    // Transaction table
                    Collection txnList = jdbcStore.getTransactions(brokerID);
                    Iterator txnItr = txnList.iterator();
                    while (txnItr.hasNext()) {
                        TransactionUID tid = (TransactionUID)txnItr.next();
                        TransactionInfo txnInfo = jdbcStore.getTransactionInfo(tid);
                        TransactionAcknowledgement txnAck[] =
                            jdbcStore.getTransactionAcks(tid);
                        bkrFS.storeTransaction(tid, txnInfo, false);
                        for (int i = 0, len = txnAck.length; i < len; i++) {
                            bkrFS.storeTransactionAck(tid, txnAck[i], false);
                        }
                    }

                    /**************************************************
                     * JMSBridge
                     **************************************************/

                    Properties bp = new Properties();
                    bp.setProperty("instanceRootDir", instanceRootDir); 
                    bp.setProperty("reset", "true");
                    bp.setProperty("logdomain", "imqdbmgr");
                    bp.setProperty("identityName", Globals.getIdentityName());


                    BridgeServiceManager.getExportedService(
                        "com.sun.messaging.bridge.service.JMSBridgeStore", "JMS", bp);

                    List bnames = jdbcStore.getJMSBridgesByBroker(brokerID, null);
                    Collections.sort(bnames);

                    String bname = null;
                    String currbname = null;
                    Iterator itr = bnames.iterator();
                    while (itr.hasNext()) {
                        bname = (String)itr.next();
                        if (currbname == null || !currbname.equals(bname)) {
                            currbname = bname;
                            bp.setProperty("jmsbridge", bname);
                            if (jmsbridgeStore != null) {
                                jmsbridgeStore.closeJMSBridgeStore();
                                jmsbridgeStore = null;
                            }
                            logger.logToAll(logger.INFO, "Backup JMS bridge "+bname);
                            jmsbridgeStore = (JMSBridgeStore)BridgeServiceManager.getExportedService(
                                             "com.sun.messaging.bridge.service.JMSBridgeStore", "JMS", bp);
                            List lrs = jdbcStore.getLogRecordsByNameByBroker(bname, brokerID, null);
                            logger.logToAll(logger.INFO, "\tBackup JMS bridge "+bname+" with "+lrs.size()+" TM log records");
                            byte[] lr = null;
                            Iterator itr1 = lrs.iterator();
                            while (itr1.hasNext()) {
                                lr = (byte[])itr1.next();
                                jmsbridgeStore.storeTMLogRecord(null, lr, currbname, true, null);
                            }
                         }
                    }

                } finally {
                    bkrFS.close();
                    if (jmsbridgeStore != null) jmsbridgeStore.closeJMSBridgeStore();
                }
            }

            logger.logToAll(Logger.INFO, "Backup persistent store complete.");
        } catch(Throwable ex){
            ex.printStackTrace();
            throw new BrokerException(ex.getMessage());
        } finally {
            if (fileStore != null) {
                try {
                    fileStore.close();
                } catch(Exception e) {
                    e.printStackTrace();
                }
            }

            StoreManager.releaseStore(true);
        }
    }

    /*
     * Restore the JDBC store from filebased backup files.
     */
    private void doRestore() throws BrokerException {

        if (!Globals.getHAEnabled()) {
            throw new BrokerException(br.getKString(BrokerResources.I_HA_NOT_ENABLE,
                dbmgr.getBrokerID()));
        }

        // instantiate the file store.
        Properties props = System.getProperties();
        String clusterID = Globals.getClusterID();
        String backupDir = (String)props.get(Globals.IMQ + ".backupdir") +
            File.separator + clusterID;

        logger.logToAll(Logger.INFO,
            "Restore persistent store for HA cluster " + clusterID + " from backup dir: " + backupDir);

        FileStore fileStore = new FileStore(backupDir, false);

        // Brokers table.
        JDBCStore jdbcStore = null;
        try {
            // Re-create the jdbc store
            doRecreate();

            jdbcStore = (JDBCStore)StoreManager.getStore();

            /*
             * For data that are not broker specific, i.e. properties, change
             * records, consumers, brokers, and sessions, we will retrieve
             * those data from the top level directory (a.k.a cluster filestore).
             */

            // Properties table
            List haBrokers = null;
            Properties properties = fileStore.getAllProperties();
            Iterator propItr = properties.entrySet().iterator();
            while (propItr.hasNext()) {
                Map.Entry entry = (Map.Entry)propItr.next();
                String name = (String)entry.getKey();
                if (name.equals(STORE_PROPERTY_HABROKERS)) {
                    // Retrieve all HABrokerInfo from a property
                    haBrokers = (List)entry.getValue();
                } else {
                    jdbcStore.updateProperty(name, entry.getValue(), false);
                }
            }
            propItr = null; properties = null;

            if (haBrokers == null || haBrokers.isEmpty()) {
                throw new BrokerException(br.getKString(
                    BrokerResources.X_LOAD_ALL_BROKERINFO_FAILED));
            }

            // Configuration Change Record table
            Object[] recordArray = fileStore.getAllConfigRecords();
            ArrayList tsList = (ArrayList)recordArray[0];
            ArrayList recList = (ArrayList)recordArray[1];
            for (int i = 0, len = tsList.size(); i < len; i++) {
                jdbcStore.storeConfigChangeRecord(
                    ((Long)tsList.get(i)).longValue(),
                    (byte[])recList.get(i), false);
            }
            recordArray = null; tsList = null; recList = null;

            // Consumer table
            Consumer[] consumerArray = fileStore.getAllInterests();
            for (int i = 0; i < consumerArray.length; i++) {
                jdbcStore.storeInterest(consumerArray[i], false);
            }
            consumerArray = null;

            // Broker & Session table.
            Iterator bkrItr = haBrokers.iterator();
            while (bkrItr.hasNext()) {
                HABrokerInfo bkrInfo = (HABrokerInfo)bkrItr.next();
                jdbcStore.addBrokerInfo(bkrInfo, false);
            }

            /*
             * For each broker in the cluster, we will retrieve broker specific
             * data, destinations, messages, transactions, acknowledgements
             * from their own filestore (a.k.a broker filestore).
             */
            bkrItr = haBrokers.iterator();
            while (bkrItr.hasNext()) {
                // Backup data for each broker
                HABrokerInfo bkrInfo = (HABrokerInfo)bkrItr.next();
                String brokerID = bkrInfo.getId();
                long sessionID = bkrInfo.getSessionID();

                logger.logToAll(Logger.INFO,
                    "Restore persistent data for broker " + brokerID);

                FileStore bkrFS = null;
                JMSBridgeStore jmsbridgeStore = null;
                try {
                    String instanceRootDir = backupDir + File.separator + brokerID;
                    bkrFS = new FileStore(instanceRootDir, false);

                    // Destination table.
                    Destination[] dstArray = bkrFS.getAllDestinations();
                    for (int i = 0, len = dstArray.length; i < len; i++) {
                        DestinationUID did = dstArray[i].getDestinationUID();
                        Destination dst = jdbcStore.getDestination(did);
                        if (dst == null) {
                            // Store the destination if not found
                            jdbcStore.storeDestination(dstArray[i], sessionID);
                        }
                    }

                    // Retrieve messages for each destination.
                    for (int i = 0, len = dstArray.length; i < len; i++) {
                        for (Enumeration e = bkrFS.messageEnumeration(dstArray[i]);
                            e.hasMoreElements();) {
                            DestinationUID did = dstArray[i].getDestinationUID();
                            Packet message = (Packet)e.nextElement();
                            SysMessageID mid = message.getSysMessageID();

                            // Get interest states for the message; Consumer State table
                            HashMap stateMap = bkrFS.getInterestStates(did, mid);
                            if (stateMap == null || stateMap.isEmpty()) {
                                jdbcStore.storeMessage(
                                    did, message, null, null, sessionID, false);
                            } else {
                                int size = stateMap.size();
                                ConsumerUID[] iids = new ConsumerUID[size];
                                int[] states  = new int[size];
                                Iterator stateItr = stateMap.entrySet().iterator();
                                int j = 0;
                                while (stateItr.hasNext()) {
                                    Map.Entry entry = (Map.Entry)stateItr.next();
                                    iids[j] = (ConsumerUID)entry.getKey();
                                    states[j] = ((Integer)entry.getValue()).intValue();
                                    j++;
                                }
                                jdbcStore.storeMessage(
                                    did, message, iids, states, sessionID, false);
                            }
                        }
                    }

                    // Transaction table
                    Collection txnList = bkrFS.getTransactions(brokerID);
                    Iterator txnItr = txnList.iterator();
                    while (txnItr.hasNext()) {
                        TransactionUID tid = (TransactionUID)txnItr.next();
                        TransactionInfo txnInfo = bkrFS.getTransactionInfo(tid);
                        TransactionAcknowledgement txnAck[] =
                            bkrFS.getTransactionAcks(tid);
                        jdbcStore.storeTransaction(tid, txnInfo, sessionID);
                        for (int i = 0, len = txnAck.length; i < len; i++) {
                            jdbcStore.storeTransactionAck(tid, txnAck[i], false);
                        }
                    }

                    /**************************************************
                     * JMSBridge
                     **************************************************/

                    Properties bp = new Properties();
                    bp.setProperty("instanceRootDir", instanceRootDir); 
                    bp.setProperty("reset", "false");
                    bp.setProperty("logdomain", "imqdbmgr");

                    jmsbridgeStore = (JMSBridgeStore)BridgeServiceManager.getExportedService(
                                     "com.sun.messaging.bridge.service.JMSBridgeStore", "JMS", bp);

                    if (jmsbridgeStore != null) {

                    List bnames = jmsbridgeStore.getJMSBridges(null);
                    String bname = null;
                    Iterator itr = bnames.iterator();
                    while (itr.hasNext()) {
                        bname = (String)itr.next();
                        jdbcStore.addJMSBridge(bname, true, null);
                    }
                    jmsbridgeStore.closeJMSBridgeStore();
                    jmsbridgeStore = null;

                    bname = null;
                    itr = bnames.iterator();
                    while (itr.hasNext()) {
                        bname = (String)itr.next();
                        bp.setProperty("jmsbridge", bname);
                        if (jmsbridgeStore != null) {
                            jmsbridgeStore.closeJMSBridgeStore();
                            jmsbridgeStore = null;
                        }
                        logger.logToAll(logger.INFO, "Restore JMS bridge "+bname);
                        jmsbridgeStore = (JMSBridgeStore)BridgeServiceManager.getExportedService(
                                         "com.sun.messaging.bridge.service.JMSBridgeStore", "JMS", bp);
                        List xids = jmsbridgeStore.getTMLogRecordKeysByName(bname, null);

                        logger.logToAll(logger.INFO, "\tRestore JMS bridge "+bname+" with "+xids.size()+" TM log records");

                        String xid = null;
                        byte[] lr = null;
                        Iterator itr1 = xids.iterator();
                        while (itr1.hasNext()) {
                            xid = (String)itr1.next();
                            lr = jmsbridgeStore.getTMLogRecord(xid, bname, null);
                            if (lr == null) {
                                logger.logToAll(Logger.INFO,
                                        "JMSBridge TM log record not found for "+xid);
                                continue;
                            }  
                            jdbcStore.storeTMLogRecord(xid, lr, bname, true, null);   
                         }
                    }
                    }

                } finally {
                    bkrFS.close();
                    if (jmsbridgeStore != null) jmsbridgeStore.closeJMSBridgeStore();
                }
            }

            logger.logToAll(Logger.INFO, "Restore persistent store complete.");
        } catch (Throwable ex){
            ex.printStackTrace();
            throw new BrokerException(ex.getMessage());
        } finally {
            if (fileStore != null) {
                try {
                    fileStore.close();
                } catch(Exception e) {
                    e.printStackTrace();
                }
            }

            StoreManager.releaseStore(true);
        }
    }

    /*
     * Check persistence store.
     */
    private void doQuery() throws BrokerException {

        Connection conn = null;
        try {
            conn = dbmgr.getConnection(true);

            if ( Globals.getHAEnabled() ) {
                System.out.println( br.getString(
                    BrokerResources.I_RUNNING_IN_HA,
                    Globals.getBrokerID(), Globals.getClusterID() ) );
            } else {
                System.out.println( br.getString(
                    BrokerResources.I_STANDALONE_INITIALIZED ) );
            }

            updateStoreVersion410IfNecessary( conn );
            int status = dbmgr.checkStoreExists( conn );
            if ( status == 0 ) {
                System.out.println( br.getString(
                    BrokerResources.E_NO_DATABASE_TABLES ) );
            } else if (status > 0) {
                // All tables have already been created
                System.out.println( br.getString(
                    BrokerResources.E_DATABASE_TABLE_ALREADY_CREATED ) );
            } else {
                // Some tables are missings
                System.out.println( br.getString(
                    BrokerResources.E_BAD_STORE_MISSING_TABLES ) );
            }
        } finally {
            Util.close( null, null, conn );
        }
    }

    private boolean printHelp(String[] args) {
	for (int i = 0; i < args.length; i++) {
	    if (args[i].equals(OPT_H) || args[i].equals(OPT_LH)) {
		return true;
	    }
	}
	return false;
    }

    private boolean printVersion(String[] args) {
	for (int i = 0; i < args.length; i++) {
	    if (args[i].equals(OPT_V) || args[i].equals(OPT_LV)) {
		return true;
	    }
	}
	return false;
    }

    // throw exception if argument is missing
    private void checkArg(String cmd, String opt, int next, int total)
	throws ParserException {
        checkArg(cmd, opt, null, next, total);
    }

    private void checkArg(String cmd, String opt, String missopt, int next, int total)
	throws ParserException {

	if (next >= total) {
	    ParserException pe;
	    if (cmd != null) {
		pe = new ParserException(MISSING_CMD_ARG);
		pe.cmd = cmd;
        } else if(missopt != null) {
		pe = new ParserException(MISSING_OPT);
		pe.opt = missopt;
	    } else {
		pe = new ParserException(MISSING_OPT_ARG);
		pe.opt = opt;
	    }

	    throw pe;
	}
    }

    private void throwParserException(int reason, String cmd, String cmdarg,
	String opt, String optarg) throws ParserException {

	ParserException pe = new ParserException(reason);
	pe.cmd = cmd;
	pe.cmdarg = cmdarg;
	pe.opt = opt;
	pe.optarg = optarg;
	throw pe;
    }

    private void handleParserException(ParserException e) {
	if (e.reason == MISSING_CMD_ARG) {
	    System.out.println(br.getString(
			BrokerResources.E_MISSING_DBMGR_CMD_ARG, e.cmd));
	} else if (e.reason == MISSING_OPT_ARG) {
	    System.out.println(br.getString(
			BrokerResources.E_MISSING_DBMGR_OPT_ARG, e.opt));
	} else if (e.reason == MISSING_OPT) {
	    System.out.println(br.getString(
			BrokerResources.E_MISSING_DBMGR_OPT, e.opt));
	} else if (e.reason == BAD_CMD_ARG) {
	    System.out.println(br.getString(
			BrokerResources.E_INVALID_DBMGR_CMD_ARG, e.cmd, e.cmdarg));
	} else if (e.reason == BAD_OPT) {
	    if (e.opt.equals(OPT_P)) {
	        System.out.println(br.getString(BrokerResources.E_PASSWD_OPTION_NOT_SUPPORTED, e.opt));
	    } else {
	        System.out.println(br.getString(BrokerResources.E_INVALID_DBMGR_OPT, e.opt));
	    }
	} else if (e.reason == BAD_OPT_ARG) {
	    System.out.println(br.getString(
			BrokerResources.E_INVALID_DBMGR_OPT_ARG, e.opt, e.optarg));
	} else if (e.reason == EXTRA_CMD_SPECIFIED) {
	    System.out.println(br.getString(
			BrokerResources.E_EXTRA_DBMGR_CMD, e.cmd));
	}
    }

    private Properties parseArgs(String[] args) throws ParserException {

	Properties props = System.getProperties();
	boolean gotcmd = false;

	// check cmd and options: -b, -p, -u (-h and -v checked already)
	for (int i = 0; i < args.length; i++) {
	    if (args[i].equals(CREATE_CMD_STR)) {
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i], null,
					null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		// find argument
		if (args[i].equals(ARGU_ALL)) {
		    props.put(CMD_NAME, CREATE_ALL_CMD);
		} else if (args[i].equals(ARGU_TBL)) {
		    props.put(CMD_NAME, CREATE_TBL_CMD);
		} else {
		    throwParserException(BAD_CMD_ARG, CREATE_CMD_STR,
					args[i], null, null);
		}
	    } else if (args[i].equals(DELETE_CMD_STR)) {
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
					null, null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		// find argument
		if (args[i].equals(ARGU_TBL)) {
		    props.put(CMD_NAME, DELETE_TBL_CMD);
		} else if (args[i].equals(ARGU_OLDTBL)) {
		    props.put(CMD_NAME, DELETE_TBL_CMD);
		    props.put(ARG_NAME, ARGU_OLDTBL);
		} else {
		    throwParserException(BAD_CMD_ARG, DELETE_CMD_STR,
					args[i], null, null);
		}
	    } else if (args[i].equals(RECREATE_CMD_STR)) {
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
					null, null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		// find argument
		if (args[i].equals(ARGU_TBL)) {
		    props.put(CMD_NAME, RECREATE_TBL_CMD);
		} else {
		    throwParserException(BAD_CMD_ARG, RECREATE_CMD_STR,
					args[i], null, null);
		}
	    } else if (args[i].equals(REMOVE_CMD_STR)) {
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
					null, null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		// find argument
		if (args[i].equals(ARGU_BKR)) {
		    props.put(CMD_NAME, REMOVE_BKR_CMD);
		} else if (args[i].equals(ARGU_JMSBRIDGE)) {
		    props.put(CMD_NAME, REMOVE_JMSBRIDGE_CMD);
		} else {
		    throwParserException(BAD_CMD_ARG, REMOVE_CMD_STR,
					args[i], null, null);
		}
            } else if (args[i].equals(DUMP_CMD)) {
		// private command for debug purpose only
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
					null, null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		props.put(CMD_NAME, DUMP_CMD);
		props.put(ARG_NAME, args[i]);
            } else if (args[i].equals(DROPTBL_CMD)) {
		// private command to drop tables that match the specified
                // table name pattern for internal use only!!!
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
					null, null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		props.put(CMD_NAME, DROPTBL_CMD);
		props.put(ARG_NAME, args[i]);
	    } else if (args[i].equals(RESET_CMD)) {
		// command for resetting the table lock
		if (gotcmd) {
		    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
					null, null, null);
		} else {
		    gotcmd = true;
		}

		checkArg(args[i], null, ++i, args.length);

		// find argument
		if (args[i].equals(ARGU_LCK)) {
		    props.put(CMD_NAME, RESET_CMD);
		} else {
		    throwParserException(BAD_CMD_ARG, RESET_CMD,
					args[i], null, null);
		}
            } else if (args[i].equals(BACKUP_CMD)) {
                // command for backing up the jdbc store
                if (gotcmd) {
                    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
                    null, null, null);
                } else {
                    gotcmd = true;
                }

                checkArg(args[i], null, ++i, args.length);

                // find argument
                if (args[i].equals(OPT_DIR)) {
                    props.put(CMD_NAME, BACKUP_CMD);
                    checkArg(args[i], null, ++i, args.length);
                    props.put(Globals.IMQ + ".backupdir", args[i]);
                } else {
                    throwParserException(BAD_CMD_ARG, BACKUP_CMD,
                    args[i], null, null);
                }
            } else if (args[i].equals(RESTORE_CMD)) {
                // command for backing up the jdbc store
                if (gotcmd) {
                    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
                    null, null, null);
                } else {
                    gotcmd = true;
                }

                checkArg(args[i], null, ++i, args.length);

                // find argument
                if (args[i].equals(OPT_DIR)) {
                    props.put(CMD_NAME, RESTORE_CMD);
                    checkArg(args[i], null, ++i, args.length);
                    props.put(Globals.IMQ + ".backupdir", args[i]);
                } else {
                    throwParserException(BAD_CMD_ARG, RESTORE_CMD,
                    args[i], null, null);
                }
            } else if (args[i].equals(UPGRADE_CMD_STR)) {
                if (gotcmd) {
                    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
                                        null, null, null);
                } else {
                    gotcmd = true;
                }

                checkArg(args[i], null, ++i, args.length);

                // find argument
                if (args[i].equals(ARGU_STORE)) {
                    props.put(CMD_NAME, UPGRADE_STORE_CMD);
                } else if (args[i].equals(ARGU_HASTORE)) {
                    props.put(CMD_NAME, UPGRADE_HASTORE_CMD);
                } else {
                    throwParserException(BAD_CMD_ARG, UPGRADE_CMD_STR,
                                        args[i], null, null);
                }
            } else if (args[i].equals(QUERY_CMD)) {
                if (gotcmd) {
                    throwParserException(EXTRA_CMD_SPECIFIED, args[i],
                                        null, null, null);
                } else {
                    gotcmd = true;
                }
                props.put(CMD_NAME, QUERY_CMD);
            } else if (args[i].equals(OPT_B)) {

                checkArg(null, OPT_B, ++i, args.length);
                props.put(Globals.IMQ + ".instancename", args[i]);

            } else if (args[i].equals(OPT_N)) {

                checkArg(null, OPT_N, ++i, args.length);
                props.put(Globals.BROKERID_PROPERTY, args[i]);
                props.put(JMSBRIDGE_NAME_PROPERTY, args[i]);

	    } else if (args[i].equals(OPT_U)) {

		checkArg(null, OPT_U, ++i, args.length);
		props.put(DBManager.FALLBACK_USER_PROP, args[i]);

	    } else if (args[i].equals(OPT_PW)) {

		checkArg(null, OPT_PW, ++i, args.length);
		props.put(DBManager.FALLBACK_PWD_PROP, args[i]);
		cliPasswdSpecified = true;

	    } else if (args[i].equals(OPT_PASSFILE)) {

		checkArg(null, OPT_PASSFILE, ++i, args.length);

		File passfile = null;
		try {
		    passfile = (new File(args[i])).getCanonicalFile();
		} catch (Exception e) {
		    throwParserException(BAD_OPT, null, null, args[i], null);
		}
		props.put(Globals.KEYSTORE_USE_PASSFILE_PROP, "true");
		props.put(Globals.KEYSTORE_PASSDIR_PROP,
					passfile.getParent());
		props.put(Globals.KEYSTORE_PASSFILE_PROP,
					passfile.getName());

            } else if (args[i].equals(OPT_VERBOSE)) {
                // Handled by wrapper script
                ;
            } else if (args[i].equals(OPT_DEBUG)) {
                debugSpecified = true;
            } else if (args[i].equals(OPT_FORCE)) {
                forceSpecified = true;
            } else if (args[i].equals(OPT_VARHOME)) {
                // Handled by wrapper script
                i++;
	    } else if (args[i].startsWith(OPT_D)) {
		int value_index = 0;
		String prop_name = null, prop_value = "";

		value_index = args[i].indexOf('=');
		if (args[i].length() <= 2) {
		    // -D
		    continue;
		} else if (value_index < 0) {
		    // -Dfoo
		    prop_name = args[i].substring(2);
		} else if (value_index == args[i].length() - 1) {
		    // -Dfoo=
		    prop_name = args[i].substring(2, value_index);
                } else {
		    // -Dfoo=bar
		    prop_name = args[i].substring(2, value_index);
		    prop_value = args[i].substring(value_index + 1);
                }

		props.put(prop_name, prop_value);
	    } else {
		throwParserException(BAD_OPT, null, null, args[i], null);
	    }
        if (props.getProperty(REMOVE_JMSBRIDGE_CMD) != null &&
            props.getProperty(JMSBRIDGE_NAME_PROPERTY) == null) {
            checkArg(null, null, OPT_N, 0, 0);
        }
	}

	return props;
    }

    /*
     * Read/parse the passfile if it is specified.
     *
     * The objective is to obtain the database password.
     * It can be specified directly via the -p option
     * or via a passfile, specified using the -passfile option.
     *
     * This method is called to to see if a passfile is specified.
     * If it was, it is read and parsed.
     *
     * Matters are complicated by the fact that imqdbmgr reads
     * the broker's configuration files. This means that
     * A password or passfile can be specified either via
     * the broker's config.properties file or imqdbmgr's
     * -p/-passfile command line options.
     *
     * A slightly less frequent use case to consider is the one
     * where the user (intentionally or not) specifies a password
     * as well as a passfile. In this case, the password takes
     * precedence over the passfile.
     *
     * Summary of behavior:
     *  - Values specified on the command line (-p, -passfile) override
     *    config.properties (imq.persist.jdbc.password, imq.passfile.*).
     *
     *  - The passwd obtained from a 'password' configuration (-p or
     *    imq.persist.jdbc.password in config.properties) will override
     *    passwords obtained from a 'passfile' configuration (-passfile
     *    or imq.passfile.* properties in config.properties). This means
     *    -p overrides -passfile.
     *
     *  - A password obtained from -passfile will override a value for
     *    imq.persist.jdbc.password in config.properties.
     */
    private void parsePassfile()
        throws FileNotFoundException
    {
        String	pf_value = null,
        	pf_dir = null,
		usePassfile = null;

	/*
	 * This method checks if a passfile was specified and merges
	 * it's contents (it should contain imq.persist.jdbc.password)
	 * with the broker configuration.
	 */

	/*
	 * Return if a passfile was not specified on the cmdline or via
	 * the broker configuration file.
	 */
        usePassfile = config.getProperty(Globals.KEYSTORE_USE_PASSFILE_PROP);
        if ((usePassfile == null) ||
	    !usePassfile.equalsIgnoreCase(Boolean.TRUE.toString())) {
	    return;
	}

	// get passfile location
        if ((pf_value = config.getProperty(Globals.KEYSTORE_PASSDIR_PROP))
                                                                != null) {
	    pf_value = StringUtil.expandVariables(pf_value, config);
	    pf_dir = pf_value;
	} else {
	    pf_dir = config.getProperty(Globals.IMQ + ".etchome")
					+File.separator + "security";
	}

	String passfile_location = pf_dir  +File.separator +
                    config.getProperty(Globals.KEYSTORE_PASSFILE_PROP);

	// Check if the passfile exists else throw exception
	File pf = new File(passfile_location);
	if (pf.exists()) {
	    Properties props = new Properties();
	    // read password from passfile
            //
	    try {
	        FileInputStream fis = new FileInputStream(
				        passfile_location);
                props.load(fis);
	    } catch (IOException ioex) {
	        logger.log(Logger.ERROR,
		    br.getKString(BrokerResources.X_READ_PASSFILE), ioex);
	    }
            config.putAll(props);
        } else {
	    throw new FileNotFoundException(
                br.getKString(BrokerResources.E_GET_PASSFILE,
                    passfile_location));
	}
    }


    private void exit(int status) {
	if (standalone) {
            DBConnectionPool.close(); // Free connection from connection pool
	    System.exit(status);
	}
    }

    void doCommand(String[] args) throws SQLException,
			BrokerException, FileNotFoundException {

	// print all
	if (args.length == 0) {
	    System.out.println(version.getBanner(true));
	    System.out.println(br.getString(BrokerResources.M_DBMGR_USAGE));
	    exit(0);
	}

	// print help
	if (printHelp(args)) {
	    System.out.println(br.getString(BrokerResources.M_DBMGR_USAGE));
	    exit(0);
	}

	// print version
	if (printVersion(args)) {
	    System.out.println(version.getBanner(true));
	    System.out.println(br.getString(BrokerResources.I_JAVA_VERSION) +
		System.getProperty("java.version") + " " +
		System.getProperty("java.vendor") + " " +
		System.getProperty("java.home"));
	    System.out.println(br.getString(BrokerResources.I_JAVA_CLASSPATH) +
		System.getProperty("java.class.path"));
	    exit(0);
	}

	Properties props = null;

	try {
            props = parseArgs(args);
	} catch (ParserException e) {
	    handleParserException(e);
	    exit(1);
	}

        // Limit the # of conn in the pool to 2
        props.setProperty( DBConnectionPool.NUM_CONN_PROP, "2" );

	if (cliPasswdSpecified)  {
            System.err.println(br.getString(BrokerResources.W_PASSWD_OPTION_DEPRECATED, OPT_PW));
            System.err.println("");
	}

        // 1st check existence of broker instance because Globals.init()
        // method will create an instance if it does not exist!
        String configName = (String)props.getProperty(Globals.IMQ + ".instancename");
        if (configName != null && configName.length() > 0) {
            Globals.pathinit(null);
            String topname =
                Globals.JMQ_INSTANCES_HOME + File.separator + configName;

            if (!(new File(topname)).exists()) {
                System.err.println(br.getString(
                    BrokerResources.E_INSTANCE_NOT_EXIST, configName));
                System.exit(1);
            }
        }

	Globals.init(props, false, false);
        config = Globals.getConfig();
	logger = Globals.getLogger();

        // Password in passfile override broker's properties object
        parsePassfile();

        // User/Password passed in from command line option take precedence.
        // Inorder to override the User/Password we need to construct the
        // correct property based on the dbVendor value; we do this after
        // calling Globals.init() so we can determine the dbVendor value.
        String dbVendor = config.getProperty(DBManager.JDBC_PROP_PREFIX + ".dbVendor");
        String vendorPropPrefix = DBManager.JDBC_PROP_PREFIX + "." + dbVendor;
        String user = props.getProperty(DBManager.FALLBACK_USER_PROP);
        if (user != null && user.length() > 0) {
            config.put(vendorPropPrefix + ".user", user);
        }

        String pwd = props.getProperty(DBManager.FALLBACK_PWD_PROP);
        if (pwd != null && pwd.length() > 0) {
            config.put(vendorPropPrefix + ".password", pwd);
        }

	dbmgr = DBManager.getDBManager();

	String cmd = props.getProperty(CMD_NAME);

	// print out info messages
	String brokerid = dbmgr.getBrokerID();
        String url;
        if (CREATE_ALL_CMD.equals(cmd)) {
            url = dbmgr.getCreateDBURL();
        } else {
            url = dbmgr.getOpenDBURL();
        }
        if ( url == null ) {
            url = "not specified";
        }
        user = dbmgr.getUser();
        if ( user == null ) {
            user = "not specified";
        }

        String msgArgs[] = { String.valueOf(JDBCStore.STORE_VERSION), brokerid,
                             url, user };
        logger.logToAll(Logger.INFO,
            br.getString(BrokerResources.I_JDBC_STORE_INFO, msgArgs));

	if (CREATE_ALL_CMD.equals(cmd)) {
	    doCreate(true); // create database as well
	} else if (CREATE_TBL_CMD.equals(cmd)) {
	    doCreate(false); // just tables
	} else if (DELETE_TBL_CMD.equals(cmd)) {
	    doDelete(props.getProperty(ARG_NAME));
        } else if (RECREATE_TBL_CMD.equals(cmd)) {
            doRecreate();
        } else if (REMOVE_BKR_CMD.equals(cmd)) {
            doRemoveBkr();
        } else if (REMOVE_JMSBRIDGE_CMD.equals(cmd)) {
            doRemoveJMSBridge(props.getProperty(JMSBRIDGE_NAME_PROPERTY));
        } else if (DUMP_CMD.equals(cmd)) {
            String arg = props.getProperty(ARG_NAME);
            doDump(arg);
        } else if (DROPTBL_CMD.equals(cmd)) {
            String arg = props.getProperty(ARG_NAME);
            doDropTablesByPattern(arg);
	} else if (RESET_CMD.equals(cmd)) {
	    doReset();
        } else if(BACKUP_CMD.equals(cmd)) {
            doBackup();
        } else if(RESTORE_CMD.equals(cmd)) {
            doRestore();
        } else if(UPGRADE_STORE_CMD.equals(cmd)) {
            doUpgrade(false);
        } else if(UPGRADE_HASTORE_CMD.equals(cmd)) {
            doUpgrade(true);
        } else if(QUERY_CMD.equals(cmd)) {
            doQuery();
	} else if (cmd == null) {
	    System.out.println(br.getString(BrokerResources.E_MISSING_DBMGR_CMD));
	} else {
	    System.out.println(br.getString(BrokerResources.E_INVALID_DBMGR_CMD, cmd));
	}
    }

    private static class ParserException extends Exception {
	String cmd;
	String cmdarg;
	String opt;
	String optarg;
	int reason;

	ParserException(int reason) {
	    this.reason = reason;
	}
    }

    public static void main(String[] args) {

        int exitCode = 0;
        DBTool tool = new DBTool(true); // standalone;
	try {
	    tool.doCommand(args);
	} catch (Exception e) {
            if (e instanceof BrokerException) {
                logger.log(Logger.ERROR, e.getMessage(), e.getCause());
            } else {
                logger.log(Logger.ERROR, e.getMessage());
            }

            if (tool.debugSpecified) {
                e.printStackTrace();
            }
	    exitCode = 1;
	} finally {
            if (tool.dbmgr != null) {
                tool.dbmgr.close(); // Free connection from connection pool
            }
        }

        System.exit(exitCode);
    }
}
