Ez\DbLinker\Driver\Connection\MasterSlavesConnection

Complexity

Total Complexity

61

Size/Duplication

Total Lines	323
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes

0

28 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	8	1
A	disableCache()	0	3	1
A	checkSlaves()	0	8	3
A	connectToMaster()	0	12	3
B	connectToSlave()	0	19	6
A	isConnectedToMaster()	0	4	2
A	getCurrentConnection()	0	4	1
A	wrap()	0	13	4
B	chooseASlave()	0	17	5
A	totalSlavesWeight()	0	8	2
A	disableCurrentSlave()	0	9	2
A	slaves()	0	4	1
A	prepare()	0	5	1
A	query()	0	7	2
A	quote()	0	4	1
A	exec()	0	5	1
A	lastInsertId()	0	5	1
A	beginTransaction()	0	5	1
A	commit()	0	5	1
A	rollBack()	0	5	1
A	errorCode()	0	4	1
A	errorInfo()	0	4	1
A	close()	0	6	2
A	hasCache()	0	3	1
A	getCacheKey()	0	3	1
A	setSlaveStatus()	0	6	2
C	getSlaveStatus()	0	25	7
B	isSlaveOk()	0	19	6

<?php

namespace Ez\DbLinker\Driver\Connection;

use Exception;
use Doctrine\DBAL\DriverManager;
use Doctrine\DBAL\Driver\Connection;
use Doctrine\DBAL\Driver\PDOConnection;

class MasterSlavesConnection implements Connection, ConnectionWrapper
{
    use ConnectionWrapperTrait;

    private $master;
    private $slaves;
    private $currentConnectionParams;
    private $currentSlave;
    private $cache;
    private $forceMaster;
    private $maxSlaveDelay = 30;
    private $slaveStatusCacheTtl = 10;

    public function __construct(array $master, array $slaves, $cache = null)
    {
        $this->master = $master;
        $this->checkSlaves($slaves);
        $this->slaves = $slaves;
        $this->cache = $cache;
        $this->forceMaster = false;
    }

    public function disableCache() {
        return $this->cache->disableCache();
    }

    private function checkSlaves(array $slaves)
    {
        foreach ($slaves as $slave) {
            if ((int)$slave['weight'] < 0) {
                throw new Exception('Slave weight must be >= 0');
            }
        }
    }

    public function connectToMaster($forced = null)
    {
        if ($forced !== null) {
            $this->forceMaster = $forced;
        }
        if ($this->currentConnectionParams === $this->master) {
            return;
        }
        $this->currentConnectionParams = $this->master;
        $this->currentSlave = null;
        $this->wrappedConnection = null;
    }

    public function connectToSlave()
    {
        $this->forceMaster = false;
        if ($this->currentConnectionParams !== null // if connection exists
            && (
                $this->currentConnectionParams !== $this->master // and is not a master
                || $this->currentConnectionParams === $this->currentSlave // or is current slave
                )
            ) {
            return;
        }
        $this->currentConnectionParams = null;
        $this->currentSlave = null;
        $this->wrappedConnection = null;
        $this->wrap();
        while (!$this->isSlaveOk() && $this->currentSlave !== null) {
            $this->wrap();
        }
    }

    public function isConnectedToMaster()
    {
        return $this->currentSlave === null && $this->currentConnectionParams !== null;
    }

    /**
     * @inherit
     */
    public function getCurrentConnection()
    {
        return $this->wrappedConnection();
    }

    protected function wrap()
    {
        if ($this->wrappedConnection !== null) {
            return $this->wrappedConnection;
        }
        if ($this->currentConnectionParams === null) {
            $this->currentSlave = $this->chooseASlave();
            $this->currentConnectionParams = $this->currentSlave !== null ? $this->slaves[$this->currentSlave] : $this->master;
        }
        $connection = DriverManager::getConnection($this->currentConnectionParams);
        $this->wrappedConnection = $connection->getWrappedConnection();
        $this->wrappedDriver = $connection->getDriver();
    }

    private function chooseASlave()
    {
        $totalSlavesWeight = $this->totalSlavesWeight();
        if ($totalSlavesWeight < 1) {
            return null;
        }
        $weightTarget = mt_rand(1, $totalSlavesWeight);
        foreach ($this->slaves as $n => $slave) {
            if ($slave['weight'] <= 0) {
                continue;
            }
            $weightTarget -= $slave['weight'];
            if ($weightTarget <= 0) {
                return $n;
            }
        }
    }

    private function totalSlavesWeight()
    {
        $weight = 0;
        foreach ($this->slaves as $slave) {
            $weight += $slave['weight'];
        }
        return $weight;
    }

    public function disableCurrentSlave()
    {
        if ($this->currentSlave !== null) {
            array_splice($this->slaves, $this->currentSlave, 1);
            $this->currentSlave = null;
        }
        $this->currentConnectionParams = null;
        $this->wrappedConnection = null;
    }

    public function slaves()
    {
        return $this->slaves;
    }

    /**
     * Prepares a statement for execution and returns a Statement object.
     *
     * @param string $prepareString
     *
     * @return \Doctrine\DBAL\Driver\Statement
     */
    public function prepare($prepareString)
    {
        $this->connectToMaster(true);
        return $this->wrappedConnection()->prepare($prepareString);
    }

    /**
     * Executes an SQL statement, returning a result set as a Statement object.
     *
     * @return \Doctrine\DBAL\Driver\Statement
     */
    public function query()
    {
        if ($this->forceMaster !== true) {
            $this->connectToSlave();
        }
        return call_user_func_array([$this->wrappedConnection(), __FUNCTION__], func_get_args());
    }

    /**
     * Quotes a string for use in a query.
     *
     * @param string  $input
     * @param integer $type
     *
     * @return string
     */
    public function quote($input, $type = \PDO::PARAM_STR)
    {
        return $this->wrappedConnection()->quote($input, $type);
    }

    /**
     * Executes an SQL statement and return the number of affected rows.
     *
     * @param string $statement
     *
     * @return integer
     */
    public function exec($statement)
    {
        $this->connectToMaster();
        return $this->wrappedConnection()->exec($statement);
    }

    /**
     * Returns the ID of the last inserted row or sequence value.
     *
     * @param string|null $name
     *
     * @return string
     */
    public function lastInsertId($name = null)
    {
        $this->forceMaster = true;
        return $this->wrappedConnection()->lastInsertId($name);
    }

    /**
     * Initiates a transaction.
     *
     * @return boolean TRUE on success or FALSE on failure.
     */
    public function beginTransaction()
    {
        $this->connectToMaster(true);
        return $this->wrappedConnection()->beginTransaction();
    }

    /**
     * Commits a transaction.
     *
     * @return boolean TRUE on success or FALSE on failure.
     */
    public function commit()
    {
        $this->connectToMaster(false);
        return $this->wrappedConnection()->commit();
    }

    /**
     * Rolls back the current transaction, as initiated by beginTransaction().
     *
     * @return boolean TRUE on success or FALSE on failure.
     */
    public function rollBack()
    {
        $this->connectToMaster(false);
        return $this->wrappedConnection()->rollBack();
    }

    /**
     * Returns the error code associated with the last operation on the database handle.
     *
     * @return string|null The error code, or null if no operation has been run on the database handle.
     */
    public function errorCode()
    {
        return $this->wrappedConnection()->errorCode();
    }

    /**
     * Returns extended error information associated with the last operation on the database handle.
     *
     * @return array
     */
    public function errorInfo()
    {
        return $this->wrappedConnection()->errorInfo();
    }

    public function close()
    {
        if (!$this->wrappedConnection() instanceof PDOConnection) {
            return $this->wrappedConnection()->getWrappedResourceHandle()->close();
        }
    }

    private function hasCache() {
        return $this->cache !== null;
    }

    private function getCacheKey() {
        return "MasterSlavesConnection_".strtr(serialize($this->currentConnectionParams), '{}()/@:', '______|');
    }

    public function setSlaveStatus(bool $running, ?int $delay) {
        if ($this->hasCache()) {
            $this->cache->setCacheItem($this->getCacheKey(), ["running" => $running, "delay" => $delay], $this->slaveStatusCacheTtl);
        }
        return ['running' => $running, 'delay' => $delay];
    }

    private function getSlaveStatus() {
        if (stripos($this->wrappedDriver->getName(), 'pgsql') !== false) {
            try {
                $sss = $this->wrappedConnection()->query("SELECT CASE WHEN pg_last_xlog_receive_location() IS NULL THEN NULL WHEN pg_last_xlog_receive_location() = pg_last_xlog_replay_location() THEN '00:00:00' ELSE now() - pg_last_xact_replay_timestamp() END AS replication_lag")->fetch();
                return $this->setSlaveStatus(true, $sss['replication_lag']);
            } catch (\Exception $e) {
                return $this->setSlaveStatus(false, null);
            }
        } else {
            try {
                $sss = $this->wrappedConnection()->query("SHOW SLAVE STATUS")->fetch();
                if ($sss['Slave_IO_Running'] === 'No' || $sss['Slave_SQL_Running'] === 'No') {
                    // slave is STOPPED
                    return $this->setSlaveStatus(false, null);
                } else {
                    return $this->setSlaveStatus(true, $sss['Seconds_Behind_Master']);
                }
            } catch (\Exception $e) {
                if (stripos($e->getMessage(), "Access denied") !== false) {
                    return $this->setSlaveStatus(true, 0);
                }
                return $this->setSlaveStatus(false, null);
            }
        }
    }

    public function isSlaveOk($maxdelay = null) {
        if ($maxdelay === null) {
            $maxdelay = $this->maxSlaveDelay;
        }
        if ($this->hasCache()) {
            $status = $this->cache->getCacheItem($this->getCacheKey());
            if ($status === null) {
                $status = $this->getSlaveStatus();
            }
        } else {
            $status = $this->getSlaveStatus();
        }
        if (!$status['running'] || $status['delay'] >= $maxdelay) {
            $this->disableCurrentSlave();
            $this->wrap();
            return false;
        }
        return true;
    }
}