Monospice\LaravelRedisSentinel\Connections\PredisConnection

Complexity

Total Complexity

18

Size/Duplication

Total Lines	229
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	12	2
A	setSentinelTimeout()	0	6	1
A	setRetryLimit()	0	7	1
A	setRetryWait()	0	7	1
A	setUpdateSentinels()	0	6	1
A	createSubscription()	0	21	4
A	transaction()	0	6	1
A	retryOnFailure()	0	19	3
A	getMaster()	0	4	1
A	getRandomSlave()	0	16	3

<?php

namespace Monospice\LaravelRedisSentinel\Connections;

use Closure;
use Illuminate\Redis\Connections\PredisConnection as LaravelPredisConnection;
use Monospice\SpicyIdentifiers\DynamicMethod;
use Predis\ClientInterface as Client;
use Predis\CommunicationException;
use RuntimeException;

/**
 * Executes Redis commands using the Predis client.
 *
 * This package extends Laravel's PredisConnection class to work around issues
 * experienced when using the Predis client to send commands over "aggregate"
 * connections (in this case, Sentinel connections).
 *
 * @category Package
 * @package  Monospice\LaravelRedisSentinel
 * @author   @pdbreen, Cy Rossignol <cy@rossignols.me>
 * @license  See LICENSE file
 * @link     https://github.com/monospice/laravel-redis-sentinel-drivers
 */
class PredisConnection extends LaravelPredisConnection
{
    /**
     * The number of times the client attempts to retry a command when it fails
     * to connect to a Redis instance behind Sentinel.
     *
     * @var int
     */
    protected $retryLimit = 20;

    /**
     * The time in milliseconds to wait before the client retries a failed
     * command.
     *
     * @var int
     */
    protected $retryWait = 1000;

    /**
     * Create a Redis Sentinel connection using a Predis client.
     *
     * @param Client $client          The Redis client to wrap.
     * @param array  $sentinelOptions Sentinel-specific connection options.
     */
    public function __construct(Client $client, array $sentinelOptions = [ ])
    {
        parent::__construct($client);

$client of type object<Predis\ClientInterface> is not a sub-type of object<Predis\Client>. It seems like you assume a concrete implementation of the interface Predis\ClientInterface to be always present.

        // Set the Sentinel-specific connection options on the Predis Client
        // connection and the current instance of this class.
        foreach ($sentinelOptions as $option => $value) {
            DynamicMethod::parseFromUnderscore($option)
                ->prepend('set')
                ->callOn($this, [ $value ]);
        }
    }

    /**
     * Set the default amount of time to wait before determining that a
     * connection attempt to a Sentinel server failed.
     *
     * @param float $seconds The timeout value in seconds.
     *
     * @return $this The current instance for method chaining.
     */
    public function setSentinelTimeout($seconds)
    {
        $this->client->getConnection()->setSentinelTimeout($seconds);

It seems like you code against a concrete implementation and not the interface Predis\Connection\ConnectionInterface as the method setSentinelTimeout() does only exist in the following implementations of said interface: Predis\Connection\Aggregate\SentinelReplication.

        return $this;
    }

    /**
     * Set the default number of attempts to retry a command when the client
     * fails to connect to a Redis instance behind Sentinel.
     *
     * @param int $attempts With a value of 0, throw an exception after the
     * first failed attempt. Pass a value of -1 to retry connections forever.
     *
     * @return $this The current instance for method chaining.
     */
    public function setRetryLimit($attempts)
    {
        $this->retryLimit = (int) $attempts;
        $this->client->getConnection()->setRetryLimit($attempts);

It seems like you code against a concrete implementation and not the interface Predis\Connection\ConnectionInterface as the method setRetryLimit() does only exist in the following implementations of said interface: Predis\Connection\Aggregate\RedisCluster, Predis\Connection\Aggregate\SentinelReplication.

        return $this;
    }

    /**
     * Set the time to wait before retrying a command after a connection
     * attempt failed.
     *
     * @param int $milliseconds The wait time in milliseconds. When 0, retry
     * a failed command immediately.
     *
     * @return $this The current instance for method chaining.
     */
    public function setRetryWait($milliseconds)
    {
        $this->retryWait = (int) $milliseconds;
        $this->client->getConnection()->setRetryWait($milliseconds);

It seems like you code against a concrete implementation and not the interface Predis\Connection\ConnectionInterface as the method setRetryWait() does only exist in the following implementations of said interface: Predis\Connection\Aggregate\SentinelReplication.

        return $this;
    }

    /**
     * Set whether the client should update the list of known Sentinels each
     * time it needs to connect to a Redis server behind Sentinel.
     *
     * @param bool $enable If TRUE, fetch the updated Sentinel list.
     *
     * @return $this The current instance for method chaining.
     */
    public function setUpdateSentinels($enable)
    {
        $this->client->getConnection()->setUpdateSentinels($enable);

It seems like you code against a concrete implementation and not the interface Predis\Connection\ConnectionInterface as the method setUpdateSentinels() does only exist in the following implementations of said interface: Predis\Connection\Aggregate\SentinelReplication.

        return $this;
    }

    /**
     * Subscribe to a set of given channels for messages.
     *
     * @param  array|string $channels The names of the channels to subscribe to.
     * @param  Closure      $callback Executed for each message. Receives the
     * message string in the first argument and the message channel as the
     * second argument.
     * @param  string       $method   The subscription command ("subscribe" or
     * "psubscribe").
     *
     * @return void
     */
    public function createSubscription(
        $channels,
        Closure $callback,
        $method = 'subscribe'
    ) {
        // Messages published to the master propagate to each of the slaves. We
        // pick a random slave to distribute load away from the master:
        $loop = $this->retryOnFailure(function () {
            return $this->getRandomSlave()->pubSubLoop();
        });

        call_user_func_array([$loop, $method], (array) $channels);

        foreach ($loop as $message) {
            if ($message->kind === 'message' || $message->kind === 'pmessage') {
                call_user_func($callback, $message->payload, $message->channel);
            }
        }

        unset($loop);
    }

    /**
     * Execute commands in a transaction.
     *
     * This package overrides the transaction() method to work around a
     * limitation in the Predis API that disallows transactions on "aggregate"
     * connections like Sentinel. Note that transactions execute on the Redis
     * master instance.
     *
     * @param callable $callback Contains the Redis commands to execute in the

Should the type for parameter $callback not be null|callable? Also, consider making the array more specific, something like array<String>, or String[].

     * transaction. The callback receives a Predis\Transaction\MultiExec
     * transaction abstraction as the only argument. We use this object to
     * execute Redis commands by calling its methods just like we would with
     * the Laravel Redis service.
     *
     * @return array|Predis\Transaction\MultiExec An array containing the
     * result for each command executed during the transaction. If no callback
     * provided, returns an instance of the Predis transaction abstraction.
     */
    public function transaction(callable $callback = null)
    {
        return $this->retryOnFailure(function () use ($callback) {
            return $this->getMaster()->transaction($callback);
        });
    }

    /**
     * Attempt to retry the provided operation when the client fails to connect
     * to a Redis server.
     *
     * We adapt Predis' Sentinel connection failure handling logic here to
     * reproduce the high-availability mode provided by the actual client. To
     * work around "aggregate" connection limitations in Predis, this class
     * provides methods that don't use the high-level Sentinel connection API
     * of Predis directly, so it needs to handle connection failures itself.
     *
     * @param callable $callback The operation to execute.
     *
     * @return mixed The result of the first successful attempt.
     */
    protected function retryOnFailure(callable $callback)
    {
        $attempts = 0;

        do {
            try {
                return $callback();
            } catch (CommunicationException $exception) {
                $exception->getConnection()->disconnect();
                $this->client->getConnection()->querySentinel();

It seems like you code against a concrete implementation and not the interface Predis\Connection\ConnectionInterface as the method querySentinel() does only exist in the following implementations of said interface: Predis\Connection\Aggregate\SentinelReplication.

                usleep($this->retryWait * 1000);

                $attempts++;
            }
        } while ($attempts <= $this->retryLimit);

        throw $exception;
    }

    /**
     * Get a Predis client instance for the master.
     *
     * @return Client The client instance for the current master.

Consider making the return type a bit more specific; maybe use \Predis\Client.

     */
    protected function getMaster()
    {
        return $this->client->getClientFor('master');
    }

    /**
     * Get a Predis client instance for a random slave.
     *
     * @param bool $fallbackToMaster If TRUE, return a client for the master
     * if the connection does not include any slaves.
     *
     * @return Client The client instance for the selected slave.
     */
    protected function getRandomSlave($fallbackToMaster = true)
    {
        $slaves = $this->client->getConnection()->getSlaves();

It seems like you code against a concrete implementation and not the interface Predis\Connection\ConnectionInterface as the method getSlaves() does only exist in the following implementations of said interface: Predis\Connection\Aggregate\MasterSlaveReplication, Predis\Connection\Aggregate\SentinelReplication.

        if (count($slaves) > 0) {
            $slave = $slaves[rand(1, count($slaves)) - 1];

            return $this->client->getClientFor($slave->getParameters()->alias);
        }

        if ($fallbackToMaster) {
            return $client->getClientFor('master');

The variable $client does not exist. Did you forget to declare it?

        }

        throw new RuntimeException('No slave present on connection.');
    }
}