The big reshuffle

1 files changed, 220 insertions, 0 deletions

diff --git a/project2/sql/rdbmsDataSource.cpp b/project2/sql/rdbmsDataSource.cpp
new file mode 100644
index 0000000..dac6cb1
--- /dev/null
+++ b/project2/sql/rdbmsDataSource.cpp
@@ -0,0 +1,220 @@
+#include "rdbmsDataSource.h"
+#include "connectionLoader.h"
+#include <libxml++/nodes/textnode.h>
+#include <sys/utsname.h>
+#include "logger.h"
+#include <errno.h>
+#include <boost/foreach.hpp>
+
+SimpleMessageException(UnknownConnectionProvider);
+
+/// Specialized ElementLoader for instances of RdbmsDataSource; handles persistent DB connections
+class RdbmsDataSourceLoader : public ElementLoaderImpl<RdbmsDataSource> {
+	public:
+		void onIdle()
+		{
+			// Disconnect all cached database connections
+			RdbmsDataSource::dbhosts.clear();
+		}
+		static bool isConnectionExpired(const RdbmsDataSource::DBHosts::value_type & con)
+		{
+			return con.second->isExpired();
+		}
+		void onPeriodic()
+		{
+			// Disconnect expired database connections
+			RdbmsDataSource::DBHosts::iterator i;
+			while ((i = std::find_if(RdbmsDataSource::dbhosts.begin(), RdbmsDataSource::dbhosts.end(), isConnectionExpired)) != RdbmsDataSource::dbhosts.end()) {
+				RdbmsDataSource::dbhosts.erase(i);
+			}
+		}
+		void onIteration()
+		{
+			RdbmsDataSource::changedDSNs.clear();
+		}
+};
+DECLARE_CUSTOM_LOADER("rdbmsdatasource", RdbmsDataSourceLoader);
+
+RdbmsDataSource::DBHosts RdbmsDataSource::dbhosts;
+RdbmsDataSource::FailedHosts RdbmsDataSource::failedhosts;
+RdbmsDataSource::DSNSet RdbmsDataSource::changedDSNs;
+
+RdbmsDataSource::RdbmsDataSource(const xmlpp::Element * p) :
+	DataSource(p),
+	masterDsn(dynamic_cast<const xmlpp::Element *>(p->find("masterdsn").front())),
+	preferLocal(p->get_attribute_value("preferlocal") != "false")
+{
+	BOOST_FOREACH(const xmlpp::Node * node, p->find("readonly/dsn")) {
+		const xmlpp::Element * elem = dynamic_cast<const xmlpp::Element *>(node);
+		if (elem) {
+			roDSNs.insert(ReadonlyDSNs::value_type(elem->get_attribute_value("host"), elem));
+		}
+	}
+}
+
+RdbmsDataSource::~RdbmsDataSource()
+{
+}
+
+void
+RdbmsDataSource::loadComplete(const CommonObjects *)
+{
+}
+
+const DB::Connection &
+RdbmsDataSource::getWritable() const
+{
+	ConnectionPtr master = connectTo(masterDsn);
+	if (!master->txOpen) {
+		master->connection->beginTx();
+		master->txOpen = true;
+	}
+	changedDSNs.insert(name);
+	return *master->connection;
+}
+
+const DB::Connection &
+RdbmsDataSource::getReadonly() const
+{
+	if (changedDSNs.find(name) != changedDSNs.end()) {
+		return *connectTo(masterDsn)->connection;
+	}
+	if (localhost.length() == 0 && preferLocal) {
+		struct utsname name;
+		if (uname(&name)) {
+			Logger()->messagef(LOG_WARNING, "%s: Unable to determine local host name (%d:%s)",
+					__PRETTY_FUNCTION__, errno, strerror(errno));
+			localhost = "unknown";
+		}
+		else {
+			localhost = name.nodename;
+		}
+	}
+	if (preferLocal) {
+		ReadonlyDSNs::const_iterator ro = roDSNs.find(localhost);
+		try {
+			if (ro == roDSNs.end()) {
+				Logger()->messagef(LOG_INFO, "%s: No database host matches local host name (%s) Will use master DSN",
+						__PRETTY_FUNCTION__, localhost.c_str());
+				return *connectTo(masterDsn)->connection;
+			}
+			return *connectTo(ro->second)->connection;
+		}
+		catch (...) {
+			// Failed to connect to a preferred DB... carry on and try the others...
+		}
+	}
+	BOOST_FOREACH(ReadonlyDSNs::value_type db, roDSNs) {
+		try {
+			return *connectTo(db.second)->connection;
+		}
+		catch (...) {
+		}
+	}
+	return *connectTo(masterDsn)->connection;
+}
+
+void
+RdbmsDataSource::commit()
+{
+	DBHosts::const_iterator m = dbhosts.find(masterDsn);
+	if (m != dbhosts.end() && m->second->txOpen) {
+		m->second->connection->commitTx();
+		m->second->txOpen = false;
+	}
+}
+
+void
+RdbmsDataSource::rollback()
+{
+	DBHosts::const_iterator m = dbhosts.find(masterDsn);
+	if (m != dbhosts.end() && m->second->txOpen) {
+		m->second->connection->rollbackTx();
+		m->second->txOpen = false;
+	}
+	changedDSNs.erase(name);
+}
+
+RdbmsDataSource::ConnectionPtr
+RdbmsDataSource::connectTo(const ConnectionInfo & dsn)
+{
+	FailedHosts::iterator dbf = failedhosts.find(dsn);
+	if (dbf != failedhosts.end()) {
+		if (time(NULL) - 20 > dbf->second.FailureTime) {
+			failedhosts.erase(dbf);
+		}
+		else {
+			throw dbf->second;
+		}
+	}
+	DBHosts::const_iterator dbi = dbhosts.find(dsn);
+	if (dbi != dbhosts.end()) {
+		try {
+			dbi->second->connection->ping();
+			dbi->second->touch();
+			return dbi->second;
+		}
+		catch (...) {
+			// Connection in failed state
+			Logger()->messagef(LOG_DEBUG, "%s: Cached connection failed", __PRETTY_FUNCTION__);
+		}
+	}
+	
+	try {
+		ConnectionPtr db = ConnectionPtr(new RdbmsConnection(dsn.connect(), 300));
+		dbhosts[dsn] = db;
+		db->touch();
+		return db;
+	}
+	catch (const DB::ConnectionError & e) {
+		failedhosts.insert(FailedHosts::value_type(dsn, e));
+		throw;
+	}
+}
+
+RdbmsDataSource::RdbmsConnection::RdbmsConnection(const DB::Connection * con, time_t kat) :
+	connection(con),
+	txOpen(false),
+	lastUsedTime(0),
+	keepAliveTime(kat)
+{
+}
+
+RdbmsDataSource::RdbmsConnection::~RdbmsConnection()
+{
+	connection->finish();
+	delete connection;
+}
+
+void
+RdbmsDataSource::RdbmsConnection::touch() const
+{
+	time(&lastUsedTime);
+}
+
+bool
+RdbmsDataSource::RdbmsConnection::isExpired() const
+{
+	return (time(NULL) > lastUsedTime + keepAliveTime);
+}
+
+RdbmsDataSource::ConnectionInfo::ConnectionInfo(const xmlpp::Element * n)
+{
+	BOOST_FOREACH(const xmlpp::Node * node, n->get_children()) {
+		typeId = LoaderBase::getLoader<ConnectionLoader, UnknownConnectionProvider>(node->get_name());
+		dsn = dynamic_cast<const xmlpp::Element *>(node)->get_child_text()->get_content();
+	}
+}
+
+DB::Connection *
+RdbmsDataSource::ConnectionInfo::connect() const
+{
+	return typeId->connect(dsn);
+}
+
+bool
+RdbmsDataSource::ConnectionInfo::operator<(const RdbmsDataSource::ConnectionInfo & other) const
+{
+	return ((typeId < other.typeId) || ((typeId == other.typeId) && (dsn < other.dsn)));
+}
+