/*
 * $Source: /opt/cvsroot/erserver/src/com/postgres/replic/server/ReplicationServer.java,v $
 * $Author: ronz $ $Revision: 1.4 $ $Date: 2003/12/29 22:04:28 $
 *
 */
package com.postgres.replic.server;

import com.postgres.replic.server.props.ConnectionPoolProps;
import com.postgres.replic.server.props.ServerProps;
import com.postgres.util.Logger.LogRepl;
import com.postgres.util.Logger.Logger;
import com.postgres.util.cmd.CmdLine;
import com.postgres.util.config.ConfiguratorIntf;
import com.postgres.util.config.ReplicationConfig;
import com.postgres.util.jdbc.ConnectionPool;
import com.postgres.util.timer.TimerListener;

import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.util.Properties;

/**
 *  Description of the Class
 *
 */
public class ReplicationServer implements TimerListener, RServSignalHandler {

	private final static String APP_NAME = "java com.postgres.replic.server.ReplicationServer";
	private boolean clean;
	private CleanLogRunnable cleanRun;
	private ConfiguratorIntf config;
	private ConnectionPoolProps connPoolProps;
	private boolean debug;
	private final static String[] flags = {
		"-debug", // extra debugging
		"-verbose", // verbose
		"-replic", // execute replication thread only
		"-clean", // execute clean thread only
		"-threads", // start threads ( w/o this flag it will execute thread once)
		"-recover", // recover dead threads
	};

	private final static String[] labels = {"-rmiport"};
	private ConnectionPool masterPool;
	private boolean masterThreadState;
	private boolean recover;
	private int recoveryInterval = 30;
	private boolean replic;
	private String replicHome;
	private boolean[] replicThreadState;
	private ReplicationRunnable[] replicThreads;
	private ServerProps serverProps;

	private boolean shutDown = false;
	private ConnectionPool[] slavePools;
	private boolean threads;
	private com.postgres.util.timer.Timer timer;
	private boolean verbose = true;

	/**
	 * <p>Called by the timer each time a clock cycle expires.
	 * This gives us the opportunity to timeout connections
	 *
	 * @param  object  Parameter
	 */
	public synchronized void TimerEvent(Object object) {
		try {
			if (replic && !shutDown) {
				createReplicationThreads();
			}
		} catch (Exception e) {
		}

		try {
			if (clean && !shutDown) {
				createCleanThread();
			}
		} catch (Exception e) {
		}

	}

	/**
	 *  Description of the Method
	 *
	 * @param  args  Parameter
	 */
	public static void main(String args[]) {
		ReplicationServer replicServer = new ReplicationServer();
		replicServer.init(args);
		replicServer.run();
	}

	/**
	 *  Description of the Method
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void createCleanThread() throws ReplicationException {
		if (slavePools == null || masterPool == null) {
			throw new ReplicationException("ReplicationServer::createCleanThread: slavePools == null || masterPool == null");
		}
		if (cleanRun != null && cleanRun.isAlive()) {
			return;
		}

		try {
			cleanRun = new CleanLogRunnable(serverProps,
					connPoolProps,
					masterPool,
					slavePools[0]);
			cleanRun.setDebug(debug);
			//cleanThread = new RservThread(cleanRun);
			getLogger().info("STARTING CLEANUP THREAD");
			cleanRun.start();
		} catch (Exception e) {
			throw new ReplicationException("ReplicationServer::createCleanThread: "
					+ e.toString());
		} finally {

		}
	}

	/**
	 *  Description of the Method
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void createConnPools() throws ReplicationException {
		if (connPoolProps == null) {
			throw new ReplicationException("ReplicationServer::createConnPools: connPoolProps == null");
		}

		try {
			Properties masterProps = connPoolProps.getMasterProperties();
			Properties[] slaveProps = connPoolProps.getSlaveProperties();
			if (slaveProps.length == 0) {
				throw new ReplicationException("ReplicationServer::createConnPools: slaveProps.length == 0");
			}

			if (debug) {
				getLogger().debug("ReplicationServer::createConnPools: Got masterProps & [] slaveProps");
			}
			if (masterPool == null) {
				masterPool = new ConnectionPool();
				if (debug) {
					getLogger().debug("ReplicationServer::createConnPools: CREATED MASTER CONNECTION POOL");
				}

				masterPool.initialize(masterProps);
				masterPool.setLogger(getLogger());
				if (debug) {
					getLogger().debug("ReplicationServer::createConnPools: INITIALIZED MASTER CONNECTION POOL");
				}
			}

			if (slavePools == null) {
				slavePools = new ConnectionPool[slaveProps.length];
				for (int i = 0; i < slaveProps.length; i++) {
					if (debug) {
						getLogger().debug("ReplicationServer::createConnPools: trying to CREATE SLAVE CONNECTION POOL #" + i);
					}
					slavePools[i] = new ConnectionPool();
					if (debug) {
						getLogger().debug("ReplicationServer::createConnPools: CREATED SLAVE CONNECTION POOL #" + i);
					}

					slavePools[i].initialize(slaveProps[i]);
					slavePools[i].setLogger(getLogger());
					if (debug) {
						getLogger().debug("ReplicationServer::createConnPools: INITIALIZED SLAVE CONNECTION POOL #" + i);
					}
				}
			}
		} catch (Exception e) {
			getLogger().error("ReplicationServer::createConnPools: Cannot create pool: "
					+ e.toString());
		} finally {

		}
	}


	/**
	 *  Description of the Method
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void createReplicationThreads() throws ReplicationException {
		if (slavePools == null || masterPool == null ||
				slavePools.length == 0) {
			throw new ReplicationException("ReplicationServer::createReplicationThreads: slavePools == null || masterPool == null || slavePools.length == 0");
		}

		try {
			if (replicThreads == null) {
				replicThreads = new ReplicationRunnable[slavePools.length];
				//replicRun = new ReplicationRunnable[slavePools.length];
			}

			boolean[] replicTreadState = getReplicThreadState();
			if (!masterThreadState) {
				resetMasterPool();
			}

			for (int i = 0; i < slavePools.length; i++) {
				if (!replicTreadState[i]) {
					if (debug) {
						getLogger().debug("ReplicationServer::createReplicationThreads: before spawnReplicationThread #" + i);
					}

					resetSlavePool(i);
					replicThreads[i] = new ReplicationRunnable(serverProps,
							connPoolProps,
							masterPool,
							slavePools[i]);
					replicThreads[i].setDebug(debug);
					//replicThreads[i] = new RservThread(replicRun[i]);
					if (debug) {
						getLogger().debug("ReplicationServer::createReplicationThreads: before replicThreads[" +
								i + "].start()");
					}
					getLogger().info("STARTING REPLICATION THREAD " + i);
					replicThreads[i].start();
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
			throw new ReplicationException("ReplicationServer::createReplicationThreads: "
					+ e.toString());
		} finally {

		}
	}

	/**
	 *  Gets the logger attribute
	 *
	 * @return    The logger value
	 */
	private Logger getLogger() {
		return LogRepl.getInstance();
	}

	/*
    */
	/**
	 *  Gets the replicThreadState attribute
	 *
	 * @return                The replicThreadState value
	 * @exception  Exception  Description of the Exception
	 */
	private boolean[] getReplicThreadState() throws Exception {
		boolean allThreadsDead = true;
		if (slavePools == null || slavePools.length == 0) {
			throw new Exception("getReplicThreadState: you may not call this method when slavePools == null || slavePools.length == 0 ");
		} else {
			if (replicThreadState == null) {
				replicThreadState = new boolean[slavePools.length];
			}

			for (int i = 0; i < slavePools.length; i++) {
				if (replicThreads[i] == null || !replicThreads[i].isAlive()) {
					replicThreadState[i] = false;
				} else {
					replicThreadState[i] = true;
					if (allThreadsDead) {
						allThreadsDead = false;
					}// at least 1 thread is alive
				}
			}

			masterThreadState = !allThreadsDead;

		}
		return replicThreadState;
	}

	/**
	 *  Description of the Method
	 *
	 * @param  args  Parameter
	 * @return       Return Value
	 */
	private boolean init(String[] args) {
		boolean rc = false;
		CmdLine cline = null;
		int port = 1099;

		if (args == null || args.length <= 0) {
			//System.out.println("args == null || args.length");
			// In no flags set - use default - all true:
			replic = true;
			clean = true;
			threads = true;
			recover = true;

		}
		try {
			cline = CmdLine.instance(args, labels, flags);
			// Flags:
			verbose = verbose ? verbose : cline.getFlag("-verbose");
			debug = debug ? debug : cline.getFlag("-debug");
			replic = cline.getFlag("-replic") | replic;

			String s = cline.getParam("-rmiport");
			if (s != null) {
				if (s.equals("disabled")) {
					port = -1;
				} else {
					try {
						port = Integer.parseInt(s);
					} catch (Exception ex) {
						port = -1;
					}
				}
			}

			// disable cleanlog temporarily:
			// clean = false;
			clean = cline.getFlag("-clean") | clean;
			threads = cline.getFlag("-threads") | threads;
			recover = cline.getFlag("-recover") | recover;

			if (!replic && !clean) {
				throw new Exception("You must specify at least one of the options: [-replic] [-clean]");
			}

			if (verbose) {
				System.out.println("verbose=" + verbose);
				System.out.println("debug=" + debug);
				System.out.println("replic=" + replic);
				System.out.println("clean=" + clean);
				System.out.println("threads=" + threads);
				System.out.println("recover=" + recover);
				System.out.println("rmiport=" + port);
			}
			rc = true;
		} catch (Exception e) {
			System.out.println("ReplicationServer::init: " + e.toString());
			String usage = CmdLine.getUsage(APP_NAME, labels, flags);
			System.out.println(usage);
		}

		if (port != -1) {
			try {
				System.err.println("Starting signal handler on port " + port);
				Registry registry = LocateRegistry.createRegistry(port);
				RServSignalHandlerImpl signalHandler = new RServSignalHandlerImpl(this);
				registry.bind(RServSignalHandler.REGISTRY_LOOKUP, signalHandler);
			} catch (java.rmi.AlreadyBoundException abx) {
				System.err.println("RMI service already bound. Skipping.");
			} catch (RemoteException rx) {
				rx.printStackTrace();
				System.err.println("Error creating remote service.");
			}
		} else {
			System.err.println("Error creating remote service. Disabled from command line.");
		}

		return rc;
	}

	/**
	 *  Description of the Method
	 */
	private void initializeHome() {
		// try -Dreplic=replic.home option
		replicHome = System.getProperty("replic.home");
		if (replicHome == null || replicHome.trim().equals("")) {
			System.out.println("replic.home has not been initialized!");
			System.exit(2);
		} else {
			System.out.println("replic.home=" + replicHome);
		}
	}

	/**
	 *
	 *
	 * @exception  Exception  Description of the Exception
	 */
	private void initializeLogger() throws Exception {
		if (replicHome == null) {
			throw new ReplicationException("ReplicationServer::intitializeLogger: replicHome == null");
		}
		LogRepl.initialize(replicHome);
	}


	/**
	 *
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void initializeProps() throws ReplicationException {
		try {// try -Dreplic.home option
			if (replicHome == null || replicHome.trim().equals("")) {
				throw new ReplicationException("ReplicationServer::initializeProps: replicHome is empty");
			}

			config = ReplicationConfig.getInstance();
			config.initialize();// use default config file "replication.cfg"
			config.setDebug(debug);

			connPoolProps = new ConnectionPoolProps(config, false);
			if (debug) {
				connPoolProps.printMasterProps();
			}

			if (debug) {
				connPoolProps.printSlaveProps();
			}

			serverProps = new ServerProps(config, false);
			if (debug) {
				serverProps.printProperties();
			}

			debug = serverProps.getDebug();
			verbose = serverProps.getVerbose();

			if (verbose) {
				//config.printProps(null);
				connPoolProps.printMasterProps();
				connPoolProps.printSlaveProps();
				serverProps.printProperties();
			}
		} catch (Exception e) {
			throw new ReplicationException("ReplicationServer::initializeProps: "
					+ e.toString());
		} finally {

		}
	}

	/**
	 *
	 */
	private void postInit() {
		// Start our timer so we can timeout connections. The
		// clock cycle will be recoveryInterval seconds
		if (recover) {
			getLogger().info("ReplicationServer::run: STARTING TIMER FOR RECOVERY OPTION");
			timer = new com.postgres.util.timer.Timer(this, recoveryInterval);
			timer.start();
		}
	}

	/**
	 *
	 */
	private void resetMasterPool() {
		if (masterPool == null) {
			return;
		} else {
			try {
				resetPool(connPoolProps.getMasterProperties(), masterPool);
			} catch (Exception e) {
				getLogger().error("ReplicationServer::resetMasterPool" + e.toString());
			}
		}
	}

	/**
	 *
	 *
	 * @param  props                     Parameter
	 * @param  cPool                     Parameter
	 * @return                           Return Value
	 */
	private boolean resetPool(Properties props, ConnectionPool cPool)  {
			//throws ReplicationException {
		boolean rc = false;
		try {
			cPool.destroy();
			cPool.initialize(props);
			rc = true;
		} catch (Exception e) {
			getLogger().error("ReplicationServer::resetPool: Cannot reset pool: "
					+ e.toString());
			rc = false;
		} finally {

		}
		return rc;
	}

	/**
	 *
	 *
	 * @param  i  Parameter
	 */
	private void resetSlavePool(int i) {
		if (slavePools == null || slavePools[i] == null) {
			return;
		} else {
			try {
				resetPool(connPoolProps.getSlaveProperties()[i], slavePools[i]);
			} catch (Exception e) {
				getLogger().error("ReplicationServer::resetSlavePool" + e.toString());
			}
		}
	}

	/**
	 *  Main processing method for the ReplicationServer object
	 */
	private void run() {
		boolean loggerStarted = false;
		try {
			initializeHome();
			initializeLogger();
			initializeProps();
			loggerStarted = true;
			Runtime.getRuntime().addShutdownHook(new ShutdownHandler());
			createConnPools();
			startProcesses();
			postInit();
		} catch (Exception e) {
			if (loggerStarted) {
				System.out.println("ReplicationServer::run: " + e.toString());
			} else {
				getLogger().error("ReplicationServer::run: ", e);
			}
		} finally {

		}
	}

	/*
    */
	/**
	 *
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void runCleanThread() throws ReplicationException {
		if (slavePools == null || masterPool == null) {
			throw new ReplicationException("ReplicationServer::runCleanThread: slavePools == null || masterPool == null");
		}

		if (cleanRun != null && cleanRun.isAlive()) {
			return;
		}

		try {
			cleanRun = new CleanLogRunnable(serverProps,
					connPoolProps,
					masterPool,
					slavePools[0]);
			cleanRun.setDebug(debug);
			//cleanThread = new RservThread(cleanRun);
			getLogger().info("STARTING CLEANUP THREAD");
			//cleanThread.start();
			cleanRun.run();
		} catch (Exception e) {
			throw new ReplicationException("ReplicationServer::runCleanThread: "
					+ e.toString());
		} finally {

		}
	}


	/*
    */
	/**
	 *
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void runReplicationThreads() throws ReplicationException {
		if (slavePools == null || masterPool == null ||
				slavePools.length == 0) {
			throw new ReplicationException("ReplicationServer::runReplicationThreads: slavePools == null || masterPool == null || slavePools.length == 0");
		}

		try {
			if (replicThreads == null) {
				replicThreads = new ReplicationRunnable[slavePools.length];
				//replicRun = new ReplicationRunnable[slavePools.length];
			}

			for (int i = 0; i < slavePools.length; i++) {
				if (replicThreads[i] == null || !replicThreads[i].isAlive()) {
					if (debug) {
						getLogger().debug("ReplicationServer::runReplicationThreads: before spawnReplicationThread #" + i);
					}
					//replicThreads[i] = spawnReplicationThread(slavePools[i]);
					replicThreads[i] = new ReplicationRunnable(serverProps,
							connPoolProps,
							masterPool,
							slavePools[i]);
					replicThreads[i].setDebug(debug);
					//replicThreads[i] = new RservThread(replicRun[i]);
					if (debug) {
						getLogger().debug("ReplicationServer::runReplicationThreads: before replicRun[" +
								i + "].start()");
					}
					getLogger().info("STARTING REPLICATION RUN " + i);
					//replicThreads[i].start();
					replicThreads[i].run();
				}
			}
		} catch (Exception e) {
			throw new ReplicationException("ReplicationServer::runReplicationThreads: "
					+ e.toString());
		} finally {

		}
	}

	/**
	 *
	 *
	 * @param  slavePools                Parameter
	 * @return                           Return Value
	 * @exception  ReplicationException  Description of the Exception
	 */
	/*private Thread spawnReplicationThread(ConnectionPool slavePools)
			throws ReplicationException {
		ReplicationRunnable replicRun = new ReplicationRunnable(serverProps,
				connPoolProps,
				masterPool,
				slavePools);
		replicRun.setDebug(debug);
		replicRun.setVerbose(verbose);
		return new Thread(replicRun);
	}*/

	/**
	 *
	 *
	 * @exception  ReplicationException  Description of the Exception
	 */
	private void startProcesses()
			throws ReplicationException {
		if (replic) {
			if (threads) {
				createReplicationThreads();
			} else {
				runReplicationThreads();
			}
		}

		if (clean) {
			if (threads) {
				createCleanThread();
			} else {
				runCleanThread();
			}
		}
	}

	public void signal(int signal) {


	}


	private class ShutdownHandler extends Thread {

		boolean exit = false;

		public ShutdownHandler(boolean b) {
			exit = b;
		}

		public ShutdownHandler() {
			this(false);
		}

		/**
		 *  Main processing method for the ShutdownHandler object
		 */
		public void run() {

			try {
				if (!shutDown) {
					shutdown();
				}
			} catch (Exception xcp) {
				getLogger().log(xcp);
			}

			if (exit) {
				System.exit(0);
			}
		}
	}

	/** Receive a signal from the RServSignalHandler
	 * @param signal 	Signal to process
	 * @see com.postgres.replic.server.RServSignalHandler
	 */
	public void receiveSignal(int signal) throws RemoteException {
		switch (signal) {
			case RServSignalHandler.RSERV_SHUTDOWN:
				new ShutdownHandler(true).start();
				break;
			case RServSignalHandler.RSERV_RELOAD:
			case RServSignalHandler.RSERV_RESET:
				for (int i = 0; i < replicThreads.length; i++) {
					if (replicThreads[i] != null) {
						replicThreads[i].resetDBProcessors();
					}
				}
				break;
		}
	}

	private void shutdown() throws Exception {

		getLogger().warn("SHUTDOWN");
		shutDown = true;

		getLogger().warn("SHUTDOWN WAIT");
		getLogger().info("Terminate all the threads & database connections...");

		// Stop all the threads:
		if (replicThreads != null && replicThreads.length > 0) {
			getLogger().info("Terminate all the Replication threads...");
			for (int i = 0; i < replicThreads.length; i++) {
				replicThreads[i].interrupt();
				replicThreads[i].join();
			}
		}

		if (cleanRun != null) {
			getLogger().info("Terminate all the Cleanup threads...");
			cleanRun.interrupt();
			cleanRun.join();
		}

		if (timer != null) {
			try {
				//timer.stop();
				timer.interrupt();
			} catch (Exception e) {
			}
		}

		getLogger().warn("SHUTDOWN DONE");
	}
}

