/*
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 2001,2009 Oracle.  All rights reserved.
 *
 */

#include <unistd.h>
#include <fcntl.h>
#include <iostream>
#include <string>
#include <sys/stat.h> // for mkdir

#include <errno.h>
#include "RepNode.hpp"

using namespace DbXml;
using namespace std;

RepNode::RepNode(u_int32_t startPolicy, const string& envHome,
	const string& host, int port, u_int32_t priority,
	vector<RepNode *>& group, const Config& config)
	: startPolicy_(startPolicy),
	  envHome_(envHome),
	  host_(host),
	  port_(port),
	  priority_(priority),
	  group_(group),
	  config_(config),
	  containerName_("DbXmlHATest.dbxml"),
	  dbenv_(NULL),
	  mgr_(NULL),
	  cont_(NULL),
	  envFlags_(DB_CREATE | DB_RECOVER | DB_INIT_LOCK | DB_INIT_LOG |
	  	    DB_INIT_MPOOL | DB_INIT_TXN | DB_INIT_REP | DB_THREAD),
	  cacheSize_(10 * 1024 * 1024)
{
	id_ = group_.size();
	group_.push_back(this);
}

RepNode::~RepNode()
{
	stop();
	// erase the nodes in group vector.
	for (size_t i = 0; i < group_.size(); i++) {
		if (group_[i] == this) {
			group_.erase(group_.begin()+i);
			break;
		}
	}
}

bool RepNode::equal(const RepNode& node)
{
	if (id_ == node.id_)
		return true;
	return false;
}

void RepNode::clearDbEnv()
{
	// Clear the existing environment. Transaction is needless in this part.
	if (access(envHome_.c_str(), 0) < 0) {
		mkdir(envHome_.c_str(), 0755);
	} else {
		// Remove Container
		{
			DB_ENV* dbenv;
			createDbEnv(&dbenv, 0);
			dbenv->open(dbenv, envHome_.c_str(), envFlags_, 0);

			XmlManager mgr(dbenv, DBXML_ADOPT_DBENV);
			try {
				mgr.removeContainer(containerName_);
			} catch (...) {}
		}

		// Remove Environment
		DB_ENV* dbEnvRemove;
		createDbEnv(&dbEnvRemove, 0);
		dbEnvRemove->remove(dbEnvRemove, envHome_.c_str(), DB_FORCE);
	}
}

void RepNode::openContainer()
{
	// reopening a container is forbidden.
	if (!cont_.isNull())
		throwException("Attempt to re-open a container.");

	XmlContainerConfig conf;

	// wait for being elected as a Master or start up done(as a client).
	while (!(isMaster() || appData_.isStartupDone()))
		yield();

	if (isMaster())
		conf.setAllowCreate(true);

	XmlTransaction txn;
	while (true) {
		try {
			txn = mgr_->createTransaction();
			cont_ = mgr_->openContainer(txn, containerName_, conf);
			txn.commit();
			return;
		/*
		 * One thing to watch out for is a case where the databases
		 * you are trying to open do not yet exist. This can happen
		 * for replicas where the databases are being opened
		 * read-only. If this happens, ENOENT is returned by the
		 * open() call.
		 */
		} catch (XmlException &xe) {
			ostringstream o;
			XmlException::ExceptionCode ec = xe.getExceptionCode();
			if (ec == XmlException::DATABASE_ERROR) {
				int dbErr = xe.getDbErrno();
				if (dbErr != ENOENT && dbErr != DB_REP_LOCKOUT) {
					o << envHome_ << ":" << xe.what() << endl
					  << "DB_ERRNO:" << dbErr << endl;
					throwException(o.str());
				}
			} else if (ec != XmlException::CONTAINER_NOT_FOUND) {
				o << envHome_ << ":" << xe.what() << endl
				  << "ExceptionCode :" << ec << endl;
				throwException(o.str());
			}
			yield();
		}
	}
}

void RepNode::start()
{
	// Restart is forbidden.
	if (dbenv_ != NULL)
		throwException("Attempt to restart a running node.");

	appData_.reset();
	init();
	mgr_ = new XmlManager(dbenv_, DBXML_ADOPT_DBENV);
	openContainer();
}

void RepNode::start(u_int32_t startPolicy)
{
	startPolicy_ = startPolicy;
	start();
}

void RepNode::stop()
{
	/*
	 * 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_->log_flush(dbenv_, NULL);

	cont_ = NULL;
	if (mgr_ != NULL) {
		delete mgr_;
		mgr_ = 0;
	}

	// dbenv is deleted by XmlManager
	dbenv_ = NULL;
}

void RepNode::init()
{
	// If the node is running, just do nothing.
	if (dbenv_ != NULL)
		return;
	/*
	 * Now we open our environment handle and set the APP_DATA structure
	 * to it's app_private member.
	 */
	createDbEnv(&dbenv_, 0);
	dbenv_->app_private = &appData_;
	/*
	 * Configure the environment handle. Here we configure
	 * asynchronous transactional commits for performance reasons.
	 */
	if (config_.verbose_)
		dbenv_->set_errfile(dbenv_, stderr);
	else
		dbenv_->set_errfile(dbenv_, NULL);
	dbenv_->set_errpfx(dbenv_, envHome_.c_str());
	dbenv_->set_event_notify(dbenv_, eventCallback);
	dbenv_->repmgr_set_ack_policy(dbenv_, DB_REPMGR_ACKS_QUORUM);

	/*
	 * Configure the local address. This is the local hostname and
	 * port that this replication participant will use to receive
	 * incoming replication messages. Note that this can be performed
	 * only once for the application. It is required.
	 */
	dbenv_->repmgr_set_local_site(dbenv_, host_.c_str(), port_, 0);
	
	/*
	* Add a site to the list of replication environments known to
	* this application.
	*/
	for (size_t i = 0; i < group_.size(); i++) {
		const RepNode& node = *group_[i];
		if (!equal(node)) {
			dbenv_->repmgr_add_remote_site(dbenv_,
						       node.host_.c_str(),
						       node.port_, NULL, 0);
		}
	}
	/*
	 * Identify the number of sites in the replication group. This is
	 * necessary so that elections and permanent message handling can
	 * be performed correctly.
	 */
	if (group_.size() > 0)
		dbenv_->rep_set_nsites(dbenv_, group_.size());

	/*
	 * Set this application's priority. This is used for elections.
	 *
	 * Set this number to a positive integer, or 0 if you do not want
	 * this site to be able to become a master.
	 */
	dbenv_->rep_set_priority(dbenv_, priority_);

	/*
	 * We can now open our environment, although we're not ready to
	 * begin replicating.  However, we want to have a dbenv around
	 * so that we can send it into any of our message handlers.
	 */
	dbenv_->set_cachesize(dbenv_, 0, cacheSize_, 0);
	dbenv_->set_flags(dbenv_, DB_TXN_NOSYNC, 1);

	// Setup the verbose flag if it's required.
	if (config_.verbose_)
		dbenv_->set_verbose(dbenv_, DB_VERB_REPLICATION, 1);

	// Open dbenv and start the replication framework such that it uses 3 threads.
	dbenv_->open(dbenv_, envHome_.c_str(), envFlags_, 0);
	dbenv_->repmgr_start(dbenv_, 3, startPolicy_);
}

XmlContainer& RepNode::getContainer()
{
	if (dbenv_ == NULL && mgr_ == NULL && cont_.isNull()) {
		// The DbEnv, Manager and Container should be created/open;
		createDbEnv(&dbenv_, 0);
		dbenv_->set_errfile(dbenv_, stderr);
		dbenv_->set_errpfx(dbenv_, envHome_.c_str());
		dbenv_->set_cachesize(dbenv_, 0, cacheSize_, 0);
		dbenv_->open(dbenv_, envHome_.c_str(), envFlags_, 0);

		mgr_ = new XmlManager(dbenv_, DBXML_ADOPT_DBENV);
		cont_ = mgr_->openContainer(containerName_);
	}
	return cont_;
}

void RepNode::eventCallback(DB_ENV* dbenv, u_int32_t which, void *info)
{
	AppData *app = (AppData *)dbenv->app_private;
	switch (which) {
	case DB_EVENT_REP_MASTER:
		app->setMaster(true);
		break;

	case DB_EVENT_REP_CLIENT:
		app->setMaster(false);
		break;

	case DB_EVENT_REP_STARTUPDONE:
		app->setStartupDone(true);
		break;

	// Don't care about this one, for now.
	case DB_EVENT_REP_ELECTED:
	case DB_EVENT_REP_NEWMASTER:
	case DB_EVENT_REP_PERM_FAILED:
		break;

	default:
		dbenv->errx(dbenv, "ignoring event %d", which);
	}
}

void RepNode::createDbEnv(DB_ENV** pDbEnv, u_int32_t flag)
{
	int ret = db_env_create(pDbEnv, flag);
	if (ret != 0) {
		string o = string("Environment open failed: ") +
			   db_strerror(ret);
		throwException(o);
	}
}

void RepNode::yield()
{
	usleep(10000);
}

void RepNode::throwIfUnExpected(XmlException &xe)
{
	if (xe.getExceptionCode() != XmlException::DATABASE_ERROR ||
	    (xe.getDbErrno() != DB_LOCK_DEADLOCK &&
	     xe.getDbErrno() != DB_REP_HANDLE_DEAD &&
	     xe.getDbErrno() != DB_REP_LOCKOUT)) {

		ostringstream o;
		o << envHome_ << ":" << xe.what() << endl;
		if (xe.getExceptionCode() == XmlException::DATABASE_ERROR)
			o << "DB_ERRNO:" << xe.getDbErrno() << endl;

		printMsg(o);
		throw xe;
	}
}

void RepNode::throwException(const string& msg)
{
	string what = envHome_ + ":" + msg;
	throwException(what);
}

OutputMutex RepNode::printMutex = OutputMutex();
void RepNode::printMsg(const std::string& str)
{
	printMutex.lock();
	cout << envHome_ << ": " << str << endl;
	printMutex.unlock();
}

void RepNode::printMsg(ostringstream& oss)
{
	printMsg(oss.str());
	oss.str("");		// reset stringstream
}