ShardingConnection::executeUpdate() - Code Metrics - Inspection of "Adding connection factory and registry" - b2pweb/bdf-prime - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( 7a6bdd...324ce1 )

by Sébastien
created 2020-06-12 13:37 UTC
ShardingConnection::executeUpdate() A

↳ Parent: ShardingConnection
Complexity

Conditions	2
Paths	2
Size

Total Lines	9
Code Lines	4
Duplication

Lines	0
Ratio	0 %
Code Coverage

Tests	5
CRAP Score	2
Importance

Changes	1
Bugs	0	Features	0
Metric	Value
eloc	4
dl	0
loc	9
c	1
b	0
f	0
ccs	5
cts	5
cp	1
rs	10
cc	2
nc	2
nop	3
crap	2
<?php

namespace Bdf\Prime\Sharding;

use Bdf\Prime\Connection\SimpleConnection;
use Bdf\Prime\Connection\SubConnectionManagerInterface;
use Bdf\Prime\ConnectionManager;

use Bdf\Prime\Query\Compiler\Preprocessor\PreprocessorInterface;
use Bdf\Prime\Query\Contract\Query\InsertQueryInterface;
use Bdf\Prime\Query\Contract\Query\KeyValueQueryInterface;
use Bdf\Prime\Sharding\Query\ShardingInsertQuery;
use Bdf\Prime\Sharding\Query\ShardingKeyValueQuery;
use Doctrine\Common\EventManager;
use Doctrine\DBAL\Cache\QueryCacheProfile;
use Doctrine\DBAL\Configuration;
use Doctrine\DBAL\Driver;
use Doctrine\DBAL\Sharding\ShardingException;
use LogicException;

/**
 * ShardingConnection
 *
 * The sharding connection is a global connection (that can be a shard server) and a collection of connection wrappers for shards.
 *
 * Those methods will be used by the global connection:
 *    ShardingConnection#prepare    Will connect on global but should execute on shard if one is selected
 *    ShardingConnection#quote
 *    ShardingConnection#errorCode
 *    ShardingConnection#errorInfo
 *    ShardingConnection#ping
 *
 * Can connect the global connection if no shard has been selected:
 *    ShardingConnection#lastInsertId
 *    ShardingConnection#lastInsertId
 *
 * Be aware!!
 * This connection does not managed the distribution key. If a SQL INSERT is executed without shard selection
 * the SQL will be executed on each shards.
 *
 * The shard to use can be auto guess if query builder is used.
 *
 * The method MultiStatement#fetchColumn change the interface (it will return a array of result)
 * All aggregate function executed on each shard will return a collection of result.
 * The merge should be done outside this class
 *
 * @example:

 * // returns the count result of the first shard
 * $connection->from('test')->count());
 *
 * // returns an array containing all count result of each shards
 * $connection->query('select count(*) from test')->fetchColumn();
 *
 * @package Bdf\Prime\Sharding

 */
class ShardingConnection extends SimpleConnection implements SubConnectionManagerInterface
{
    /**
     * All shard connections
     *
     * @var SimpleConnection[]
     */
    private $connections = [];

    
    /**
     * The shard choser
     * 
     * @var ShardChoserInterface
     */
    private $shardChoser;

    /**
     * The id of current shard. Null means all shards
     *
     * @var string
     */
    private $currentShardId;

    /**
     * The distribution key
     *
     * @var string
     */
    private $distributionKey;

    /**
     * Initializes a new instance of the Connection class.
     * 
     * Here's a shard connections configuration

     * 
     * @example
     *
     * $conn = DriverManager::getConnection([
     *    'driver' => 'pdo_mysql',
     *    'user'     => 'user',
     *    'password' => 'password',
     *    'host'     => '127.0.0.1',
     *    'dbname'   => 'basename',
     *    'distributionKey' => 'id',
     *    'shards' => [
     *      '{shardId}' => [
     *        'user'     => 'shard1',
     *        'host'     => '...',
     *      ]
     *    ]
     * ]);
     *
     * @param array                              $params       The connection parameters.

     * @param \Doctrine\DBAL\Driver              $driver       The driver to use.
     * @param \Doctrine\DBAL\Configuration|null  $config       The configuration, optional.
     * @param \Doctrine\Common\EventManager|null $eventManager The event manager, optional.
     */

    public function __construct(array $params, Driver $driver, Configuration $config = null, EventManager $eventManager = null)
    {
        if (!isset($params['shard_connections'])) {
            throw new LogicException('Sharding connection needs "shard_connections" configuration in parameters');
        }

        if (!isset($params['distributionKey'])) {
            throw new LogicException('Sharding connection needs distribution key in parameters');
        }

        $this->distributionKey = $params['distributionKey'];
        $this->shardChoser = $params['shardChoser'] ?? new ModuloChoser();

        $this->connections = $params['shard_connections'];

        parent::__construct($params, $driver, $config, $eventManager);

        $this->factory()->alias(InsertQueryInterface::class, ShardingInsertQuery::class);
        $this->factory()->/** @scrutinizer ignore-call */ alias(InsertQueryInterface::class, ShardingInsertQuery::class);
        $this->factory()->alias(KeyValueQueryInterface::class, ShardingKeyValueQuery::class);
    }

    /**
     * Get the shard ids
     *
     * @return array
     */
    public function getShardIds()
    {
        return array_keys($this->connections);
    }

    /**
     * Get the shard choser
     *
     * @return ShardChoserInterface
     */
    public function getShardChoser()
    {
        return $this->shardChoser;
    }

    /**
     * Get the distribution key
     *
     * @return string
     */
    public function getDistributionKey()
    {
        return $this->distributionKey;
    }

    /**
     * Get the current shard
     *
     * @return string
     */
    public function getCurrentShardId()
    {
        return $this->currentShardId;
    }

    /**
     * Select a shard to use.
     *
     * @param mixed $distributionValue
     *
     * @return $this
     */
    public function pickShard($distributionValue = null)
    {
        $this->useShard(
            $distributionValue !== null
                ? $this->shardChoser->pick($distributionValue, $this)

                : null
        );

        return $this;
    }

    /**
     * Use a shard
     *
     * @param string $shardId
     *
     * @return $this
     *
     * @throws ShardingException   If the shard id is not known

     */
    public function useShard($shardId = null)
    {
        if ($shardId !== null && !isset($this->connections[$shardId])) {
            throw new ShardingException('Trying to use an unknown shard id "'.$shardId.'"');
        }

        $this->currentShardId = $shardId;

        return $this;
    }

    /**
     * Check whether the connection is using a shard
     *
     * @return boolean

     */
    public function isUsingShard()
    {
        return $this->currentShardId !== null;
    }

    /**
     * Get a shard connection by its id
     * Returns all connection if id is null

     *
     * @param mixed $shardId
     *
     * @return SimpleConnection|SimpleConnection[]
     *
     * @throws ShardingException   If the shard id is not known

     */
    public function getShardConnection($shardId = null)
    {
        if ($shardId === null) {
            return $this->connections;
        }

        if (!isset($this->connections[$shardId])) {
            throw new ShardingException('Trying to get an unknown shard id "'.$shardId.'"');
        }

        return $this->connections[$shardId];
    }

    /**

     * {@inheritdoc}
     */
    public function getConnection($name)
    {
        return $this->getShardConnection($name);
    }

    /**
     * Get the selected shards
     *
     * @return SimpleConnection[]
     */
    protected function getSelectedShards()
    {
        if ($this->isUsingShard()) {
            return [$this->connections[$this->currentShardId]];
        }

        return $this->connections;
    }

    /**
     * Get the selected shard
     *
     * @return SimpleConnection
     */
    protected function getSelectedShard()
    {
        return $this->connections[$this->currentShardId];
    }

    /**
     * {@inheritdoc}
     */
    public function close()
    {
        parent::close();

        $this->currentShardId = null;

        foreach ($this->connections as $shard) {
            $shard->close();
        }
    }

    /**

     * {@inheritdoc}
     */
    public function builder(PreprocessorInterface $preprocessor = null)
    {
        return $this->factory()->make(ShardingQuery::class, $preprocessor);
    }

    /**

     * {@inheritdoc}
     */
    public function executeQuery($sql, array $params = [], $types = [], QueryCacheProfile $qcp = null)
    {
        if ($this->isUsingShard()) {
            return $this->getSelectedShard()->executeQuery($sql, $params, $types, $qcp);
        }

        $stmt = new MultiStatement();

        foreach ($this->getSelectedShards() as $shard) {
            $stmt->add($shard->executeQuery($sql, $params, $types, $qcp));
        }

        return $stmt;
    }

    /**

     * {@inheritdoc}
     */
    public function executeUpdate($sql, array $params = [], array $types = [])
    {
        $result = 0;

        foreach ($this->getSelectedShards() as $shard) {
            $result += $shard->executeUpdate($sql, $params, $types);
        }

        return $result;
    }

    /**
     * {@inheritdoc}
     */
    public function query()
    {
        $args = func_get_args();

        if ($this->isUsingShard()) {
            return $this->getSelectedShard()->query(...$args);
        }

        $stmt = new MultiStatement();

        foreach ($this->getSelectedShards() as $shard) {
            $stmt->add($shard->query(...$args));
        }

        return $stmt;
interface HasName {
    /** @return string */
    public function getName();
}

class Name {
    public $name;
}

class User implements HasName {
    /** @return string|Name */
    public function getName() {
        return new Name('foo'); // This is a violation of the ``HasName`` interface
                                // which only allows a string value to be returned.
    }
}
    }

    /**

     * {@inheritdoc}
     */
    public function exec($statement)
    {
        $result = 0;

        foreach ($this->getSelectedShards() as $shard) {
            $result += $shard->exec($statement);
        }

        return $result;
    }

    /**
     * {@inheritdoc}
     */
    public function beginTransaction()
    {
        foreach ($this->getSelectedShards() as $shard) {
            $shard->beginTransaction();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function commit()
    {
        foreach ($this->getSelectedShards() as $shard) {
            $shard->commit();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function rollBack()
    {
        foreach ($this->getSelectedShards() as $shard) {
            $shard->rollBack();
        }
    }

    /**

     * {@inheritdoc}
     */
    public function createSavepoint($savepoint)
    {
        foreach ($this->getSelectedShards() as $shard) {
            $shard->createSavepoint($savepoint);
        }
    }

    /**

     * {@inheritdoc}
     */
    public function releaseSavepoint($savepoint)
    {
        foreach ($this->getSelectedShards() as $shard) {
            $shard->releaseSavepoint($savepoint);
        }
    }

    /**

     * {@inheritdoc}
     */
    public function rollbackSavepoint($savepoint)
    {
        foreach ($this->getSelectedShards() as $shard) {
            $shard->rollbackSavepoint($savepoint);
        }
    }

    /**

     * {@inheritdoc}
     */
    public function lastInsertId($seqName = null)
    {
        if ($this->isUsingShard()) {
            return $this->getSelectedShard()->lastInsertId($seqName);
        }

        // TODO doit on lever une exception ?

        return parent::lastInsertId($seqName);
    }

    /**
     * {@inheritdoc}
     */
    public function getWrappedConnection()
    {
        if ($this->isUsingShard()) {
            return $this->getSelectedShard()->getWrappedConnection();
        }

        // TODO doit on lever une exception ?

        return parent::getWrappedConnection();
    }
}

b2pweb / bdf-prime

Push — master ( 7a6bdd...324ce1 )

ShardingConnection::executeUpdate() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like
