ManialibDatabaseConnectionReadWriteSplit.php

<?php
/**
 * ManiaLib - Lightweight PHP framework for Manialinks
 *
 * @see         http://code.google.com/p/manialib/
 * @copyright   Copyright (c) 2009-2011 NADEO (http://www.nadeo.com)
 * @license     http://www.gnu.org/licenses/lgpl.html LGPL License 3
 * @version     $Revision: 749 $:
 * @author      $Author: baptiste33 $:
 * @date        $Date: 2012-08-23 14:58:11 +0200 (jeu., 23 août 2012) $:
 */

namespace ManiaLib\Database;

class Connection
{

  /**
	 * @var array[\ManiaLib\Database\ConnectionParams]
	 */
	static protected $connections = array();

	/**
	 * @var \ManiaLib\Database\Config
	 */
	protected $config;

	/**
	 * Master parameters
	 * @var \ManiaLib\Database\ConnectionsParams
	 */
	protected $params;

	/**
	 * @var \ManiaLib\Database\ConnectionsParams[]
	 */
	protected $slavesParams;

	/**
	 * MySQL connection ressource
	 */
	protected $connection;

	/**
	 * MySQL slave connection ressource
	 */
	protected $slaveConnection;


	protected $curentConnection;

	/**
	 * Current charset
	 * @var string
	 */
	protected $charset;

	/**
	 * Currently selected database
	 * @vat string
	 */
	protected $database;

	/**
	 * @var int
	 */
	protected $transactionRefCount;

	/**
	 * @var bool
	 */
	protected $transactionRollback;

	/**
	 * @var bool
	 */
	protected $forceUseMaster;

	/**
	 * The easiest way to retrieve a database connection. Works as a singleton,
	 * and returns a connection with the default parameters
	 * @return \ManiaLib\Database\Connection
	 */
	static function getInstance()
	{
		if(\array_key_exists('default', static::$connections))
		{
			return static::$connections['default'];
		}

		$config = Config::getInstance();

		$params = new ConnectionParams();
		$params->id = 'default';
		$params->host = $config->host;
		$params->user = $config->user;
		$params->password = $config->password;
		$params->database = $config->database;
		$params->charset = $config->charset;
		$params->persistent = $config->persistent;

		$slaveParams = array();
		foreach($config->slavesDSN as $index => $dsn)
		{
			$dsn = parse_url($dsn);

			$slaveParam = new ConnectionParams();
			$slaveParam->id = 'slave-'.$index;
			$slaveParam->host = $dsn['host'];
			$slaveParam->user = $dsn['user'];
			$slaveParam->pass = $dsn['pass'];
			$slaveParam->database = $config->database;
			$slaveParam->charset = $config->charset;

			$slaveParams[] = $slaveParam;
		}

		return static::factory($params, $slaveParams);
	}

	/**
	 * Advanced connection retrieval. You can have multiple instances of the
	 * Connection object so you can work with several MySQL servers. If you don't
	 * give any parameters, it will work as static::getInstance()
	 * @return \ManiaLib\Database\Connection
	 */
	static function factory(ConnectionParams $params, array $slaveParams = array())
	{
		if(!$params->id)
		{
			throw new Exception('ConnectionParams object has no ID');
		}
		if(!array_key_exists($params->id, static::$connections))
		{
			static::$connections[$params->id] = new static($params, $slaveParams);
		}
		return static::$connections[$params->id];
	}

	protected function __construct(ConnectionParams $paramsMaster, array $slavesParams)
	{
		$this->config = Config::getInstance();

		$this->params = $paramsMaster;


		if (!$slavesParams)
		{
			$this->setForceUseMaster();
		}
		$this->slavesParams = shuffle($slavesParams);
	}

	/**
	 *
	 * @param \ManiaLib\Database\ConnectionParams $params
	 * @throws Exception
	 */
	function connect(ConnectionParams $params)
	{
		if($params->persistent)
		{
			$connection = mysql_pconnect(
				$params->host, $params->user, $params->password, $params->ssl ? MYSQL_CLIENT_SSL : null);
		}
		else
		{
			$connection = mysql_connect(
				$params->host, $params->user, $params->password, true,
				$params->ssl ? MYSQL_CLIENT_SSL : null);
		}

		if(!$connection)
		{
			throw new Exception('Connection failed');
		}

		$this->curentConnection = $connection;

		$this->setCharset($params->charset);
		$this->select($params->database);

		return $this->curentConnection;
	}

	function connectToMaster()
	{
		if (!$this->connection)
		{
			$this->connection =  $this->connect($this->params);
		}
		return $this->connection;
	}

	function connectToASlave()
	{
		if (!$this->slaveConnection)
		{
			for ($i=0; $i<count($this->slavesParams); $i++)
			{
				try
				{
					$connection = $this->connect($this->slavesParams[$i]);
					break;
				}
				catch (\Exception $e)
				{

				}
			}
			if (!$connection)
			{
				\ManiaLib\Utils\Logger::error('Connected on master but slave required');
				$connection = $this->connectToMaster();
			}
			$this->slaveConnection = $connection;
		}
		return $this->slaveConnection;
	}

	function setCharset($charset)
	{
		if($charset != $this->charset)
		{
			$this->charset = $charset;
			if(!mysql_set_charset($charset, $this->curentConnection))
			{
				throw new Exception('Couldn\'t set charset: '.$charset);
			}
		}
	}

	function select($database)
	{
		if($database && $database != $this->database)
		{
			$this->database = $database;
			if(!mysql_select_db($this->database, $this->curentConnection))
			{
				throw new Exception(mysql_error(), mysql_errno());
			}
		}
	}

	function quote($string)
	{
		return '\''.mysql_real_escape_string($string, $this->currentConnection).'\'';
	}

	/**
	 * @return RecordSet
	 */
	function execute($query /* , sprintf style params... */)
	{
		$mtime = microtime(true);
		if(func_num_args() > 1)
		{
			$query = call_user_func_array('sprintf', func_get_args());
		}

		if ($this->useSlave($query))
		{
			$this->connectToASlave();
		}
		else
		{
			$this->connectToMaster();
		}

		$query = $this->instrumentQuery($query);

		$result = mysql_query($query, $this->currentConnection);
		if(!$result)
		{
			throw new QueryException(mysql_error().': '.$query, mysql_errno());
		}
		if($this->config->queryLog)
		{
			$mtime2 = round((microtime(true) - $mtime) * 1000);
			$message = str_pad($mtime2.' ms', 10, ' ').$query;
			\ManiaLib\Utils\Logger::info($message);
		}
		if($this->config->slowQueryLog)
		{
			$mtime2 = round((microtime(true) - $mtime) * 1000);
			if($mtime2 > $this->config->slowQueryThreshold)
			{
				$message = str_pad($mtime2.' ms', 10, ' ').$query;
				\ManiaLib\Utils\Logger::info($message);
			}
		}
		return new RecordSet($result);
	}

	/**
	 * @param string $query
	 * @return string Augmented query
	 */
	protected function instrumentQuery($query)
	{
		$bt = debug_backtrace();
		if(count($bt) < 3)
		{
			return $query;
		}
		array_shift($bt);
		array_shift($bt);
		$frame = array_shift($bt);
		$class = \ManiaLib\Utils\Arrays::get($frame, 'class');
		$type = \ManiaLib\Utils\Arrays::get($frame, 'type');
		$function = \ManiaLib\Utils\Arrays::get($frame, 'function');
		$line = \ManiaLib\Utils\Arrays::get($frame, 'line');

		$queryHeader = sprintf("/* Function: %s%s%s(), Line: %d*/ ", $class, $type, $function, $line);
		$query = $queryHeader.trim($query);
		return $query;
	}

	function delete($table, $identifier)
	{
		$criteria = array();
		foreach($identifier as $key => $value)
		{
			$criteria[] = sprintf('%s = %s', $key, $value);
		}
		$criteria = implode(' AND ', $criteria);
		$query = sprintf('DELETE FROM %s WHERE %s', $table, $criteria);
		return $this->execute($query);
	}

	function affectedRows()
	{
		return mysql_affected_rows($this->currentConnection);
	}

	function insertID()
	{
		return mysql_insert_id($this->currentConnection);
	}

	function isConnected()
	{
		return (!$this->currentConnection);
	}

	function getDatabase()
	{
		return $this->database;
	}

	/**
	 * ACID Transactions
	 * ONLY WORKS WITH INNODB TABLES !
	 *
	 * ----
	 *
	 * It handles EXPERIMENTAL (== never tested!!!) nested transactions
	 * one "BEGIN" on the first call of beginTransaction
	 * one "COMMIT" on the last call of commitTransaction (when the ref count is 1)
	 * one "ROLLBACK" on the first call of rollbackTransaction
	 */
	function beginTransaction()
	{
		if($this->transactionRollback)
		{
			throw new Exception('Transaction must be rollback\'ed!');
		}
		if($this->transactionRefCount === null)
		{
			$this->execute('BEGIN');
			$this->transactionRefCount = 1;
		}
		else
		{
			$this->transactionRefCount++;
		}
	}

	/**
	 * @see static::beginTransaction()
	 */
	function commitTransaction()
	{
		if($this->transactionRollback)
		{
			throw new Exception('Transaction must be rollback\'ed!');
		}
		if($this->transactionRefCount === null)
		{
			throw new Exception('Transaction was not previously started');
		}
		elseif($this->transactionRefCount > 1)
		{
			$this->transactionRefCount--;
		}
		elseif($this->transactionRefCount == 1)
		{
			$this->execute('COMMIT');
			$this->transactionRefCount = null;
		}
		else
		{
			throw new Exception(
				'Transaction reference counter error: '.
				print_r($this->transactionRefCount, true));
		}
	}

	/**
	 * @see static::beginTransaction()
	 */
	function rollbackTransaction()
	{
		if(!$this->transactionRollback)
		{
			$this->transactionRollback = true;
			$this->execute('ROLLBACK');
		}

		if($this->transactionRefCount > 1)
		{
			$this->transactionRefCount--;
		}
		elseif($this->transactionRefCount == 1)
		{
			$this->transactionRefCount = null;
			$this->transactionRollback = null;
		}
		else
		{
			throw new Exception(
				'Transaction reference counter error: '.
				print_r($this->transactionRefCount, true));
		}
	}

	function doTransaction($callback)
	{
		try
		{
			$this->beginTransaction();
			call_user_func($callback);
			$this->commitTransaction();
		}
		catch(\Exception $e)
		{
			$this->rollbackTransaction();
			throw $e;
		}
	}

	function setForceUseMaster($boolean = true)
	{
		$this->forceUseMaster = $boolean;
	}

	/**
	 * @return boolean
	 */
	protected function useSlave($query)
	{
		if ($this->forceUseMaster)
		{
			return false;
		}
		if ($this->transactionRefCount > 0)
		{
			return false;
		}
		//Maybe strip mysql comments
		if (preg_match('/^(\s*)SELECT/i', $query))
		{
			return true;
		}
		else
		{
			return false;
		}
	}

}

?>