PredisConnection::pubSubLoop() - Code Metrics - Inspection of "Update README" - monospice/laravel-redis-sentinel-drivers - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — 1.x ( 1267b2...d54394 )

by Cy
created 2018-05-10 10:29 UTC
PredisConnection::pubSubLoop() A

↳ Parent: PredisConnection
Complexity

Conditions	1
Paths	1
Size

Total Lines	6
Code Lines	2
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
dl	0
loc	6
rs	9.4285
c	0
b	0
f	0
cc	1
eloc	2
nc	1
nop	2
<?php

namespace Monospice\LaravelRedisSentinel;

use Closure;
use InvalidArgumentException;
use Predis\Client;
use Predis\CommunicationException;
use Predis\PubSub\Consumer as PubSub;
use RuntimeException;

/**
 * Executes Redis commands using the Predis client.
 *
 * This package extends the Predis client to work around issues experienced
 * when using the it to send commands over "aggregate" connections (in this
 * case, Sentinel connections).
 *
 * Unlike in the 2.x branch, this PredisConnection implementation extends the
 * Predis client instead of consuming it as a dependency so we don't break
 * backward compatibility in Laravel 5.3 and below.
 *
 * @category Package
 * @package  Monospice\LaravelRedisSentinel
 * @author   Cy Rossignol <[email protected]>
 * @license  See LICENSE file
 * @link     https://github.com/monospice/laravel-redis-sentinel-drivers
 */
class PredisConnection extends Client
{
    /**
     * 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;

    /**
     * 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->getConnection()->setSentinelTimeout($seconds);
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

        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->getConnection()->setRetryLimit($attempts);
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

        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->getConnection()->setRetryWait($milliseconds);
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

        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->getConnection()->setUpdateSentinels($enable);
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

        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. Return FALSE to unsubscribe.
     * @param string       $method   The subscription command ("subscribe" or
     * "psubscribe").
     *
     * @return void
     */
    public function createSubscription(
        $channels,
        Closure $callback,
        $method = 'subscribe'
    ) {
        $this->retryOnFailure(function () use ($method, $channels, $callback) {
            $loop = $this->pubSubLoop([ $method => (array) $channels ]);

            if ($method === 'psubscribe') {
                $messageKind = 'pmessage';
            } else {
                $messageKind = 'message';
            }

            $this->consumeMessages($loop, $messageKind, $callback);
/** @return stdClass|null */
function mayReturnNull() { }

function doesNotAcceptNull(stdClass $x) { }

// With potential error.
function withoutCheck() {
    $x = mayReturnNull();
    doesNotAcceptNull($x); // Potential error here.
}

// Safe - Alternative 1
function withCheck1() {
    $x = mayReturnNull();
    if ( ! $x instanceof stdClass) {
        throw new \LogicException('$x must be defined.');
    }
    doesNotAcceptNull($x);
}

// Safe - Alternative 2
function withCheck2() {
    $x = mayReturnNull();
    if ($x instanceof stdClass) {
        doesNotAcceptNull($x);
    }
}

            unset($loop);
        });
    }

    /**
     * Create a new PUB/SUB subscriber and pass messages to the callback if
     * provided.
     *
     * WARNING: Consumers created using this method are not monitored for
     * connection failures. For Sentinel support, use one of the methods
     * provided by the Laravel API instead (subscribe() and psubscribe()).
     *
     * @param array|null $options  Configures the channel(s) to subscribe to.
     * @param callable   $callback Optional callback executed for each message

     * published to the configured channel(s).
     *
     * @return \Predis\PubSub\Consumer|null A PUB/SUB context used to create
     * a subscription loop if no callback provided.
     */
    public function pubSubLoop($options = null, $callback = null)
    {
        // Messages published to the master propagate to each of the slaves. We
        // pick a random slave to distribute load away from the master:
        return $this->getRandomSlave()->pubSubLoop($options, $callback);
    }

    /**
     * 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 array|callable|null $options  Predis transaction settings OR a
     * callback to execute. When passing a callback for the first argument, do
     * not supply a second argument.
     * @param callable|null       $callback Contains the Redis commands to
     * execute in the 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($options = null, callable $callback = null)
    {
        return $this->retryOnFailure(function () use ($options, $callback) {
            if ($callback !== null) {
                return $this->getMaster()->transaction($options, $callback);
            }

            return $this->getMaster()->transaction($options);
        });
    }

    /**
     * Creates a new client instance for the specified connection ID or alias,
     * only when working with an aggregate connection (cluster, replication).
     * The new client instances uses the same options of the original one.
     *
     * @param string $connectionID Identifier of a connection.
     *
     * @return Client A Predis client instance for the specified connection.
     *
     * @throws InvalidArgumentException When the aggregate connection does not
     * contain a node that matches the specified ID.
     */
    public function getClientFor($connectionID)
    {
        if (! $connection = $this->getConnectionById($connectionID)) {
            throw new InvalidArgumentException(
                "Invalid connection ID: $connectionID."
// Instead of
$x = "foo $bar $baz";

// Better use either
$x = "foo " . $bar . " " . $baz;
$x = sprintf("foo %s %s", $bar, $baz);
            );
        }

        // Because this class extends the Predis client, we need to return an
        // instance of the base client class so that clients created for the
        // Redis servers behind Sentinel don't implement the overrides here.
        return new parent($connection, $this->options);
    }

    /**
     * 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.
     *
     * @throws CommunicationException After exhausting the allowed number of
     * attempts to reconnect.
     */
    protected function retryOnFailure(callable $callback)
    {
        $attempts = 0;

        do {
            try {
                return $callback();
            } catch (CommunicationException $exception) {
                $exception->getConnection()->disconnect();
                $this->getConnection()->querySentinel();
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

                usleep($this->retryWait * 1000);

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

        throw $exception;
    }

    /**
     * Execute the provided callback for each message read by the PUB/SUB
     * consumer.
     *
     * @param PubSub  $loop     Reads the messages published to a channel.
     * @param string  $kind     The subscribed message type ([p]message).
     * @param Closure $callback Executed for each message.
     *
     * @return void
     */
    protected function consumeMessages(PubSub $loop, $kind, Closure $callback)
    {
        foreach ($loop as $message) {
            if ($message->kind === $kind) {
                if ($callback($message->payload, $message->channel) === false) {
                    return;
                }
            }
        }
    }

    /**
     * Get a Predis client instance for the master.
     *
     * @return Client The client instance for the current master.
     */
    protected function getMaster()
    {
        return $this->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.
     *
     * @throws RuntimeException When the client cannot reach any replicas
     * (and the master if $fallbackToMaster is TRUE).
     */
    protected function getRandomSlave($fallbackToMaster = true)
    {
        $slaves = $this->getConnection()->getSlaves();
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

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

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

        if ($fallbackToMaster) {
            return $this->getMaster();
        }

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

Push — 1.x ( 1267b2...d54394 )

PredisConnection::pubSubLoop() A

Complexity

Size

Duplication

Importance

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

monospice / laravel-redis-sentinel-drivers

Push — 1.x ( 1267b2...d54394 )

PredisConnection::pubSubLoop() A

Complexity

Size

Duplication

Importance

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Duplication Side-by-Side

Filter issues like
