/*-
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 1997, 2010 Oracle and/or its affiliates.  All rights reserved.
 *
 * $Id$
 */

package db.repquote;

import java.io.FileNotFoundException;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.lang.Thread;
import java.lang.InterruptedException;

import com.sleepycat.db.*;
import db.repquote.RepConfig;

/**
 * RepQuoteExample is a simple but complete demonstration of a replicated
 * application. The application is a mock stock ticker. The master accepts a
 * stock symbol and an numerical value as input, and stores this information
 * into a replicated database; either master or clients can display the
 * contents of the database.
 * <p>
 * The options to start a given replication node are:
 * <pre>
 *   -h home (required; h stands for home directory)
 *   -l host:port (required; l stands for local)
 *   -C or M (optional; start up as client or master)
 *   -r host:port (optional; r stands for remote; any number of these may
 *      be specified)
 *   -R host:port (optional; R stands for remote peer; only one of these may
 *      be specified)
 *   -a all|quorum (optional; a stands for ack policy)
 *   -b (optional; b stands for bulk)
 *   -n nsites (optional; number of sites in replication group; defaults to 0
 *      to try to dynamically compute nsites)
 *   -p priority (optional; defaults to 100)
 *   -v (optional; v stands for verbose)
 * </pre>
 * <p>
 * A typical session begins with a command such as the following to start a
 * master:
 *
 * <pre>
 *   java db.repquote.RepQuoteExample -M -h dir1 -l localhost:6000
 * </pre>
 *
 * and several clients:
 *
 * <pre>
 *   java db.repquote.RepQuoteExample -C -h dir2
 *               -l localhost:6001 -r localhost:6000
 *   java db.repquote.RepQuoteExample -C -h dir3
 *               -l localhost:6002 -r localhost:6000
 *   java db.repquote.RepQuoteExample -C -h dir4
 *               -l localhost:6003 -r localhost:6000
 * </pre>
 *
 * <p>
 * Each process is a member of a DB replication group. The sample application
 * expects the following commands to stdin:
 * <ul>
 * <li>NEWLINE -- print all the stocks held in the database</li>
 * <li>quit -- shutdown this node</li>
 * <li>exit -- shutdown this node</li>
 * <li>stock_symbol number -- enter this stock and number into the
 * database</li>
 * </ul>
 */

public class RepQuoteExample
{
    private RepConfig appConfig;
    private RepQuoteEnvironment dbenv;
    private CheckpointThread ckpThr;
    private LogArchiveThread lgaThr;

    public static void usage()
    {
        System.err.println("usage: " + RepConfig.progname +
            " -h home -l host:port [-CM][-r host:port][-R host:port]\n" +
            "  [-a all|quorum][-b][-n nsites][-p priority][-v]");

        System.err.println(
             "\t -h home (required; h stands for home directory)\n" +
             "\t -l host:port (required; l stands for local)\n" +
             "\t -C or -M (optional; start up as client or master)\n" +
             "\t -r host:port (optional; r stands for remote; any number " +
             "of these\n" +
             "\t    may be specified)\n" +
             "\t -R host:port (optional; R stands for remote peer; only " +
             "one of\n" +
             "\t    these may be specified)\n" +
             "\t -a all|quorum (optional; a stands for ack policy)\n" +
             "\t -b (optional; b stands for bulk)\n" +
             "\t -n nsites (optional; number of sites in replication " +
             "group; defaults\n" +
             "\t    to 0 to try to dynamically compute nsites)\n" +
             "\t -p priority (optional; defaults to 100)\n" +
             "\t -v (optional; v stands for verbose)\n");

        System.exit(1);
    }

    public static void main(String[] argv)
        throws Exception
    {
        RepConfig config = new RepConfig();
        boolean isPeer;
        String tmpHost;
        int tmpPort = 0;

        /*  Extract the command line parameters. */
        for (int i = 0; i < argv.length; i++)
        {
            isPeer = false;
            if (argv[i].compareTo("-a") == 0) {
                if (i == argv.length - 1)
                    usage();
                i++;
                if (argv[i].equals("all"))
                    config.ackPolicy = ReplicationManagerAckPolicy.ALL;
                else if (!argv[i].equals("quorum"))
                    usage();
            } else if (argv[i].compareTo("-b") == 0)
                config.bulk = true;
            else if (argv[i].compareTo("-C") == 0) {
                config.startPolicy = ReplicationManagerStartPolicy.REP_CLIENT;
            } else if (argv[i].compareTo("-h") == 0) {
                if (i == argv.length - 1)
                    usage();
                /* home - a string arg. */
                i++;
                config.home = argv[i];
            } else if (argv[i].compareTo("-l") == 0) {
                if (i == argv.length - 1)
                    usage();
                /* "local" should be host:port. */
                i++;
                String[] words = argv[i].split(":");
                if (words.length != 2) {
                    System.err.println(
                        "Invalid host specification host:port needed.");
                    usage();
                }
                try {
                    tmpPort = Integer.parseInt(words[1]);
                } catch (NumberFormatException nfe) {
                    System.err.println("Invalid host specification, " +
                        "could not parse port number.");
                    usage();
                }
                config.setThisHost(words[0], tmpPort);
            } else if (argv[i].compareTo("-M") == 0) {
                config.startPolicy = ReplicationManagerStartPolicy.REP_MASTER;
            } else if (argv[i].compareTo("-n") == 0) {
                if (i == argv.length - 1)
                    usage();
                i++;
                config.totalSites = Integer.parseInt(argv[i]);
            } else if (argv[i].compareTo("-p") == 0) {
                if (i == argv.length - 1)
                    usage();
                i++;
                config.priority = Integer.parseInt(argv[i]);
            } else if (argv[i].compareTo("-R") == 0 ||
                argv[i].compareTo("-r") == 0) {
                if (i == argv.length - 1)
                    usage();
                if (argv[i].equals("-R"))
                    isPeer = true;
                i++;
                String[] words = argv[i].split(":");
                if (words.length != 2) {
                    System.err.println(
                        "Invalid host specification host:port needed.");
                    usage();
                }
                try {
                    tmpPort = Integer.parseInt(words[1]);
                } catch (NumberFormatException nfe) {
                    System.err.println("Invalid host specification, " +
                        "could not parse port number.");
                    usage();
                }
                config.addOtherHost(words[0], tmpPort, isPeer);
            } else if (argv[i].compareTo("-v") == 0) {
                config.verbose = true;
            } else {
                System.err.println("Unrecognized option: " + argv[i]);
                usage();
            }

        }

        /* Error check command line. */
        if ((!config.gotListenAddress()) || config.home.length() == 0)
            usage();

        RepQuoteExample runner = null;
        try {
            runner = new RepQuoteExample();
            runner.init(config);

            /* Sleep to give ourselves time to find a master. */
            //try {
            //    Thread.sleep(5000);
            //} catch (InterruptedException e) {}

            runner.doloop();
            runner.terminate();
        } catch (DatabaseException dbe) {
            System.err.println("Caught an exception during " +
                "initialization or processing: " + dbe);
            if (runner != null)
                runner.terminate();
        }
    } /* End main. */

    public RepQuoteExample()
        throws DatabaseException
    {
        appConfig = null;
        dbenv = null;
    }

    public int init(RepConfig config)
        throws DatabaseException
    {
        int ret = 0;
        appConfig = config;
        EnvironmentConfig envConfig = new EnvironmentConfig();
        envConfig.setErrorStream(System.err);
        envConfig.setErrorPrefix(RepConfig.progname);

        envConfig.setReplicationManagerLocalSite(appConfig.getThisHost());
        for (RepRemoteHost host = appConfig.getFirstOtherHost();
            host != null; host = appConfig.getNextOtherHost()){
            envConfig.replicationManagerAddRemoteSite(
                host.getAddress(), host.isPeer());
        }
        if (appConfig.totalSites > 0)
            envConfig.setReplicationNumSites(appConfig.totalSites);

        /* 
         * Set replication group election priority for this environment.
         * An election first selects the site with the most recent log
         * records as the new master.  If multiple sites have the most
         * recent log records, the site with the highest priority value 
         * is selected as master.
         */
        envConfig.setReplicationPriority(appConfig.priority);

        envConfig.setCacheSize(RepConfig.CACHESIZE);
        envConfig.setTxnNoSync(true);

        envConfig.setEventHandler(new RepQuoteEventHandler());

        /*
         * Set the policy that determines how master and client sites
         * handle acknowledgement of replication messages needed for
         * permanent records.  The default policy of "quorum" requires only
         * a quorum of electable peers sufficient to ensure a permanent
         * record remains durable if an election is held.  The "all" option
         * requires all clients to acknowledge a permanent replication
         * message instead.
         */
        envConfig.setReplicationManagerAckPolicy(appConfig.ackPolicy);

        /*
         * Set the threshold for the minimum and maximum time the client
         * waits before requesting retransmission of a missing message.
         * Base these values on the performance and load characteristics
         * of the master and client host platforms as well as the round
         * trip message time.
         */
        envConfig.setReplicationRequestMin(20000);
        envConfig.setReplicationRequestMax(500000);

        /*
         * Configure deadlock detection to ensure that any deadlocks
         * are broken by having one of the conflicting lock requests
         * rejected. DB_LOCK_DEFAULT uses the lock policy specified
         * at environment creation time or DB_LOCK_RANDOM if none was
         * specified.
         */
        envConfig.setLockDetectMode(LockDetectMode.DEFAULT);

        envConfig.setAllowCreate(true);
        envConfig.setRunRecovery(true);
        envConfig.setThreaded(true);
        envConfig.setInitializeReplication(true);
        envConfig.setInitializeLocking(true);
        envConfig.setInitializeLogging(true);
        envConfig.setInitializeCache(true);
        envConfig.setTransactional(true);
        envConfig.setVerboseReplication(appConfig.verbose);
        try {
            dbenv = new RepQuoteEnvironment(appConfig.getHome(), envConfig);
        } catch(FileNotFoundException e) {
            System.err.println("FileNotFound exception: " + e);
            System.err.println(
                "Ensure that the environment directory is pre-created.");
            ret = 1;
        }
        
        if (appConfig.bulk)
            dbenv.setReplicationConfig(ReplicationConfig.BULK, true);

        /*
         * Configure heartbeat timeouts so that repmgr monitors the
         * health of the TCP connection.  Master sites broadcast a heartbeat
         * at the frequency specified by the DB_REP_HEARTBEAT_SEND timeout.
         * Client sites wait for message activity the length of the 
         * DB_REP_HEARTBEAT_MONITOR timeout before concluding that the 
         * connection to the master is lost.  The DB_REP_HEARTBEAT_MONITOR 
         * timeout should be longer than the DB_REP_HEARTBEAT_SEND timeout.
         */
        dbenv.setReplicationTimeout(ReplicationTimeoutType.HEARTBEAT_SEND, 
            5000000);
        dbenv.setReplicationTimeout(ReplicationTimeoutType.HEARTBEAT_MONITOR, 
            10000000);

        /* The following base replication features may also be useful to your
         * application. See Berkeley DB documentation for more details.
         *   - Master leases: Provide stricter consistency for data reads
         *     on a master site.
         *   - Timeouts: Customize the amount of time Berkeley DB waits
         *     for such things as an election to be concluded or a master
         *     lease to be granted.
         *   - Delayed client synchronization: Manage the master site's
         *     resources by spreading out resource-intensive client 
         *     synchronizations.
         *   - Blocked client operations: Return immediately with an error
         *     instead of waiting indefinitely if a client operation is
         *     blocked by an ongoing client synchronization.
         *
         * The following repmgr features may also be useful to your
         * application.  See Berkeley DB documentation for more details.
         *  - Two-site strict majority rule - In a two-site replication
         *    group, require both sites to be available to elect a new
         *    master.
         *  - Timeouts - Customize the amount of time repmgr waits
         *    for such things as waiting for acknowledgements or attempting 
         *    to reconnect to other sites.
         *  - Site list - return a list of sites currently known to repmgr.
         */

        /* Start checkpoint and log archive support threads. */ 
        ckpThr = new CheckpointThread(dbenv);
        ckpThr.start();
        lgaThr = new LogArchiveThread(dbenv, envConfig);
        lgaThr.start();

        /* Start replication manager. */
        dbenv.replicationManagerStart(3, appConfig.startPolicy);
        
        return ret;
    }

    public int doloop()
        throws DatabaseException
    {
        Database db = null;

        for (;;)
        {
            if (db == null) {
                DatabaseConfig dbconf = new DatabaseConfig();
                dbconf.setType(DatabaseType.BTREE);
                if (dbenv.getIsMaster()) {
                    /*
                     * Open database allowing create only if this is a master
                     * database.  A client database uses polling to attempt
                     * to open the database without allowing create until
                     * it is successful.
                     *
                     * This polling logic for allowing create can be 
                     * simplified under some circumstances.  For example, if
                     * the application can be sure a database is already
                     * there, it would never need to open it allowing create.
                     */
                    dbconf.setAllowCreate(true);
                }
                dbconf.setTransactional(true);

                try {
                    db = dbenv.openDatabase
                        (null, RepConfig.progname, null, dbconf);
                } catch (java.io.FileNotFoundException e) {
                    System.err.println("no stock database available yet.");
                    if (db != null) {
                        db.close(true);
                        db = null;
                    }
                    try {
                        Thread.sleep(RepConfig.SLEEPTIME);
                    } catch (InterruptedException ie) {}
                    continue;
                }
            }

            BufferedReader stdin =
                new BufferedReader(new InputStreamReader(System.in));

            /* Listen for input, and add it to the database. */
            System.out.print("QUOTESERVER");
            if (!dbenv.getIsMaster())
                System.out.print("(read-only)");
            System.out.print("> ");
            System.out.flush();
            String nextline = null;
            try {
                nextline = stdin.readLine();
            } catch (IOException ioe) {
                System.err.println("Unable to get data from stdin");
                break;
            }
            String[] words = nextline.split("\\s");

            /* A blank line causes the DB to be dumped to stdout. */
            if (words.length == 0 ||
                (words.length == 1 && words[0].length() == 0)) {
                try {
                    if (dbenv.getInClientSync())
                        System.err.println(
    "Cannot read data during client initialization - please try again.");
                    else
                        printStocks(db);
                } catch (DeadlockException de) {
                    continue;
                } catch (DatabaseException e) {
                    /*
                     * This could be DB_REP_HANDLE_DEAD, which
                     * should close the database and continue.
                     */
                    System.err.println("Got db exception reading replication" +
                        "DB: " + e);
                    System.err.println("Expected if it was due to a dead " +
                        "replication handle, otherwise an unexpected error.");
                    db.close(true); /* Close no sync. */
                    db = null;
                    continue;
                }
                continue;
            }

            if (words.length == 1 &&
                (words[0].compareToIgnoreCase("quit") == 0 ||
                words[0].compareToIgnoreCase("exit") == 0)) {
                    dbenv.setAppFinished(true);
                    break;
            } else if (words.length != 2) {
                System.err.println("Format: TICKER VALUE");
                continue;
            }

            if (!dbenv.getIsMaster()) {
                System.err.println("Can't update client.");
                continue;
            }

            DatabaseEntry key = new DatabaseEntry(words[0].getBytes());
            DatabaseEntry data = new DatabaseEntry(words[1].getBytes());

            db.put(null, key, data);
        }
        if (db != null)
            db.close(true);
        return 0;
    }

    public void terminate()
        throws DatabaseException
    {
        /* Wait for checkpoint and log archive threads to finish. */
        try {
            lgaThr.join();
            ckpThr.join();
        } catch (Exception e1) {
            System.err.println("Support thread join failed.");
        }

        /*
         * We have used the DB_TXN_NOSYNC environment flag for improved
         * performance without the usual sacrifice of transactional durability,
         * as discussed in the "Transactional guarantees" page of the Reference
         * Guide: if one replication site crashes, we can expect the data to
         * exist at another site.  However, in case we shut down all sites
         * gracefully, we push out the end of the log here so that the most
         * recent transactions don't mysteriously disappear.
         */
        dbenv.logFlush(null);
        
        dbenv.close();
    }

    /*
     * void return type since error conditions are propogated
     * via exceptions.
     */
    private void printStocks(Database db)
        throws DeadlockException, DatabaseException
    {
        Cursor dbc = db.openCursor(null, null);

        System.out.println("\tSymbol\tPrice");
        System.out.println("\t======\t=====");

        DatabaseEntry key = new DatabaseEntry();
        DatabaseEntry data = new DatabaseEntry();
        OperationStatus ret;
        for (ret = dbc.getFirst(key, data, LockMode.DEFAULT);
            ret == OperationStatus.SUCCESS;
            ret = dbc.getNext(key, data, LockMode.DEFAULT)) {
            String keystr = new String
                (key.getData(), key.getOffset(), key.getSize());
            String datastr = new String
                (data.getData(), data.getOffset(), data.getSize());
            System.out.println("\t"+keystr+"\t"+datastr);

        }
        dbc.close();
    }

    /*
     * Implemention of EventHandler interface to handle the Berkeley DB events
     * we are interested in receiving.
     */
    private /* internal */
    class RepQuoteEventHandler extends EventHandlerAdapter {
        public void handleRepClientEvent()
        {
            dbenv.setIsMaster(false);
            dbenv.setInClientSync(true);
        }
        public void handleRepMasterEvent()
        {
            dbenv.setIsMaster(true);
            dbenv.setInClientSync(false);
        }
        public void handleRepNewMasterEvent()
        {
            dbenv.setInClientSync(true);
        }
        public void handleRepPermFailedEvent()
        {
            /*
             * Did not get enough acks to guarantee transaction
             * durability based on the configured ack policy.  This 
             * transaction will be flushed to the master site's
             * local disk storage for durability.
             */
            System.err.println(
    "Insufficient acknowledgements to guarantee transaction durability.");
        }
        public void handleRepStartupDoneEvent()
        {
            dbenv.setInClientSync(false);
        }
    }
} /* End RepQuoteEventHandler class. */

/*
 * This is a very simple thread that performs checkpoints at a fixed
 * time interval.  For a master site, the time interval is one minute
 * plus the duration of the checkpoint_delay timeout (30 seconds by
 * default.)  For a client site, the time interval is one minute.
 */
class CheckpointThread extends Thread {
    private RepQuoteEnvironment myEnv = null;

    public CheckpointThread(RepQuoteEnvironment env) {
        myEnv = env;
    }

    public void run() {
        for (;;) {
            /*
             * Wait for one minute, polling once per second to see if
             * application has finished.  When application has finished,
             * terminate this thread.
             */
            for (int i = 0; i < 60; i++) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {}
                if (myEnv.getAppFinished())
                    return;
            }

            /* Perform a checkpoint. */
            try {
                myEnv.checkpoint(null);
            } catch (DatabaseException de) {
                System.err.println("Could not perform checkpoint.");
            }
        }
    }
}

/*
 * This is a simple log archive thread.  Once per minute, it removes all but
 * the most recent 3 logs that are safe to remove according to a call to
 * DBENV->log_archive().
 *
 * Log cleanup is needed to conserve disk space, but aggressive log cleanup
 * can cause more frequent client initializations if a client lags too far
 * behind the current master.  This can happen in the event of a slow client,
 * a network partition, or a new master that has not kept as many logs as the
 * previous master.
 *
 * The approach in this routine balances the need to mitigate against a
 * lagging client by keeping a few more of the most recent unneeded logs
 * with the need to conserve disk space by regularly cleaning up log files.
 * Use of automatic log removal (DBENV->log_set_config() DB_LOG_AUTO_REMOVE 
 * flag) is not recommended for replication due to the risk of frequent 
 * client initializations.
 */
class LogArchiveThread extends Thread {
    private RepQuoteEnvironment myEnv = null;
    private EnvironmentConfig myEnvConfig = null;

    public LogArchiveThread(RepQuoteEnvironment env, 
        EnvironmentConfig envConfig) {
        myEnv = env;
        myEnvConfig = envConfig;
    }

    public void run() {
        java.io.File[] logFileList;
        int logs_to_keep = 3;
        int minlog;

        for (;;) {
            /*
             * Wait for one minute, polling once per second to see if
             * application has finished.  When application has finished,
             * terminate this thread.
             */
            for (int i = 0; i < 60; i++) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {}
                if (myEnv.getAppFinished())
                    return;
            }

            try {
                /* Get the list of unneeded log files. */
                logFileList = myEnv.getArchiveLogFiles(false);
                /* 
                 * Remove all but the logs_to_keep most recent unneeded
                 * log files. 
                 */
                minlog = logFileList.length - logs_to_keep;
                for (int i = 0; i < minlog; i++) {
                    logFileList[i].delete();
                }
            } catch (DatabaseException de) {
                System.err.println("Problem deleting log archive files.");
            }
        }
    }
}