Predis\Connection\PhpiredisSocketConnection

Complexity

Total Complexity

59

Size/Duplication

Total Lines	370
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

19 Methods

Rating	Name	Duplication	Size	Complexity
A	getErrorHandler()	0	11	2
B	connectWithTimeout()	0	33	8
A	getAddress()	0	11	3
A	read()	0	19	6
A	connect()	0	8	5
A	writeRequest()	0	6	1
A	assertExtensions()	0	11	3
A	__construct()	0	7	1
A	assertParameters()	0	19	5
B	setSocketOptions()	0	28	7
A	__wakeup()	0	4	1
A	getStatusHandler()	0	11	2
A	__destruct()	0	5	1
A	createReader()	0	8	1
A	emitSocketError()	0	8	1
A	write()	0	16	4
A	getReader()	0	3	1
A	disconnect()	0	7	2
A	createResource()	0	25	5

<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2023 Till Krüss
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Connection;

use Closure;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\Status as StatusResponse;

/**
 * This class provides the implementation of a Predis connection that uses the
 * PHP socket extension for network communication and wraps the phpiredis C
 * extension (PHP bindings for hiredis) to parse the Redis protocol.
 *
 * This class is intended to provide an optional low-overhead alternative for
 * processing responses from Redis compared to the standard pure-PHP classes.
 * Differences in speed when dealing with short inline responses are practically
 * nonexistent, the actual speed boost is for big multibulk responses when this
 * protocol processor can parse and return responses very fast.
 *
 * For instructions on how to build and install the phpiredis extension, please
 * consult the repository of the project.
 *
 * The connection parameters supported by this class are:
 *
 *  - scheme: it can be either 'redis', 'tcp' or 'unix'.
 *  - host: hostname or IP address of the server.
 *  - port: TCP port of the server.
 *  - path: path of a UNIX domain socket when scheme is 'unix'.
 *  - timeout: timeout to perform the connection (default is 5 seconds).
 *  - read_write_timeout: timeout of read / write operations.
 *
 * @see http://github.com/nrk/phpiredis
 * @deprecated 2.1.2
 */
class PhpiredisSocketConnection extends AbstractConnection
{
    private $reader;

    /**
     * {@inheritdoc}
     */
    public function __construct(ParametersInterface $parameters)
    {
        $this->assertExtensions();

        parent::__construct($parameters);

        $this->reader = $this->createReader();
    }

    /**
     * Disconnects from the server and destroys the underlying resource and the
     * protocol reader resource when PHP's garbage collector kicks in.
     */
    public function __destruct()
    {
        parent::__destruct();

        phpiredis_reader_destroy($this->reader);

The function phpiredis_reader_destroy was not found. Maybe you did not declare it correctly or list all dependencies?

    }

    /**
     * Checks if the socket and phpiredis extensions are loaded in PHP.
     */
    protected function assertExtensions()
    {
        if (!extension_loaded('sockets')) {
            throw new NotSupportedException(
                'The "sockets" extension is required by this connection backend.'
            );
        }

        if (!extension_loaded('phpiredis')) {
            throw new NotSupportedException(
                'The "phpiredis" extension is required by this connection backend.'
            );
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function assertParameters(ParametersInterface $parameters)
    {
        switch ($parameters->scheme) {
            case 'tcp':
            case 'redis':
            case 'unix':
                break;

            default:
                throw new InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
        }

        if (isset($parameters->persistent)) {
            throw new NotSupportedException(
                'Persistent connections are not supported by this connection backend.'
            );
        }

        return $parameters;
    }

    /**
     * Creates a new instance of the protocol reader resource.
     *
     * @return resource
     */
    private function createReader()
    {
        $reader = phpiredis_reader_create();

The function phpiredis_reader_create was not found. Maybe you did not declare it correctly or list all dependencies?

        phpiredis_reader_set_status_handler($reader, $this->getStatusHandler());

The function phpiredis_reader_set_status_handler was not found. Maybe you did not declare it correctly or list all dependencies?

        phpiredis_reader_set_error_handler($reader, $this->getErrorHandler());

The function phpiredis_reader_set_error_handler was not found. Maybe you did not declare it correctly or list all dependencies?

        return $reader;
    }

    /**
     * Returns the underlying protocol reader resource.
     *
     * @return resource
     */
    protected function getReader()
    {
        return $this->reader;
    }

    /**
     * Returns the handler used by the protocol reader for inline responses.
     *
     * @return Closure
     */
    protected function getStatusHandler()
    {
        static $statusHandler;

        if (!$statusHandler) {
            $statusHandler = function ($payload) {
                return StatusResponse::get($payload);
            };
        }

        return $statusHandler;
    }

    /**
     * Returns the handler used by the protocol reader for error responses.
     *
     * @return Closure
     */
    protected function getErrorHandler()
    {
        static $errorHandler;

        if (!$errorHandler) {
            $errorHandler = function ($errorMessage) {
                return new ErrorResponse($errorMessage);
            };
        }

        return $errorHandler;
    }

    /**
     * Helper method used to throw exceptions on socket errors.
     */
    private function emitSocketError()
    {
        $errno = socket_last_error();
        $errstr = socket_strerror($errno);

        $this->disconnect();

        $this->onConnectionError(trim($errstr), $errno);
    }

    /**
     * Gets the address of an host from connection parameters.
     *
     * @param ParametersInterface $parameters Parameters used to initialize the connection.
     *
     * @return string
     */
    protected static function getAddress(ParametersInterface $parameters)
    {
        if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP)) {
            return $host;
        }

        if ($host === $address = gethostbyname($host)) {
            return false;

The expression return false returns the type false which is incompatible with the documented return type string.

        }

        return $address;
    }

    /**
     * {@inheritdoc}
     */
    protected function createResource()
    {
        $parameters = $this->parameters;

        if ($parameters->scheme === 'unix') {
            $address = $parameters->path;
            $domain = AF_UNIX;
            $protocol = 0;
        } else {
            if (false === $address = self::getAddress($parameters)) {

The condition false === $address = self::getAddress($parameters) is always false.

                $this->onConnectionError("Cannot resolve the address of '$parameters->host'.");
            }

            $domain = filter_var($address, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6) ? AF_INET6 : AF_INET;
            $protocol = SOL_TCP;
        }

        if (false === $socket = @socket_create($domain, SOCK_STREAM, $protocol)) {
            $this->emitSocketError();
        }

        $this->setSocketOptions($socket, $parameters);

It seems like $socket can also be of type Socket; however, parameter $socket of Predis\Connection\PhpiredisSocketConnection::setSocketOptions() does only seem to accept resource, maybe add an additional type check?

        $this->connectWithTimeout($socket, $address, $parameters);

It seems like $socket can also be of type Socket; however, parameter $socket of Predis\Connection\PhpiredisSocketConnection::connectWithTimeout() does only seem to accept resource, maybe add an additional type check?

        return $socket;
    }

    /**
     * Sets options on the socket resource from the connection parameters.
     *
     * @param resource            $socket     Socket resource.
     * @param ParametersInterface $parameters Parameters used to initialize the connection.
     */
    private function setSocketOptions($socket, ParametersInterface $parameters)
    {
        if ($parameters->scheme !== 'unix') {
            if (!socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1)) {
                $this->emitSocketError();
            }

            if (!socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1)) {
                $this->emitSocketError();
            }
        }

        if (isset($parameters->read_write_timeout)) {
            $rwtimeout = (float) $parameters->read_write_timeout;
            $timeoutSec = floor($rwtimeout);
            $timeoutUsec = ($rwtimeout - $timeoutSec) * 1000000;

            $timeout = [
                'sec' => $timeoutSec,
                'usec' => $timeoutUsec,
            ];

            if (!socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout)) {
                $this->emitSocketError();
            }

            if (!socket_set_option($socket, SOL_SOCKET, SO_RCVTIMEO, $timeout)) {
                $this->emitSocketError();
            }
        }
    }

    /**
     * Opens the actual connection to the server with a timeout.
     *
     * @param resource            $socket     Socket resource.
     * @param string              $address    IP address (DNS-resolved from hostname)
     * @param ParametersInterface $parameters Parameters used to initialize the connection.
     *
     * @return void
     */
    private function connectWithTimeout($socket, $address, ParametersInterface $parameters)
    {
        socket_set_nonblock($socket);

        if (@socket_connect($socket, $address, (int) $parameters->port) === false) {
            $error = socket_last_error();

            if ($error != SOCKET_EINPROGRESS && $error != SOCKET_EALREADY) {
                $this->emitSocketError();
            }
        }

        socket_set_block($socket);

        $null = null;
        $selectable = [$socket];

        $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);
        $timeoutSecs = floor($timeout);
        $timeoutUSecs = ($timeout - $timeoutSecs) * 1000000;

        $selected = socket_select($selectable, $selectable, $null, $timeoutSecs, $timeoutUSecs);

$timeoutSecs of type double is incompatible with the type integer expected by parameter $tv_sec of socket_select().

$null of type null is incompatible with the type array expected by parameter $except of socket_select().

$timeoutUSecs of type double is incompatible with the type integer expected by parameter $tv_usec of socket_select().

        if ($selected === 2) {
            $this->onConnectionError('Connection refused.', SOCKET_ECONNREFUSED);
        }

        if ($selected === 0) {
            $this->onConnectionError('Connection timed out.', SOCKET_ETIMEDOUT);
        }

        if ($selected === false) {
            $this->emitSocketError();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        if (parent::connect() && $this->initCommands) {

The expression $this->initCommands of type array is implicitly converted to a boolean; are you sure this is intended? If so, consider using ! empty($expr) instead to make it clear that you intend to check for an array without elements.

            foreach ($this->initCommands as $command) {
                $response = $this->executeCommand($command);

                if ($response instanceof ErrorResponseInterface) {
                    $this->onConnectionError("`{$command->getId()}` failed: {$response->getMessage()}", 0);
                }
            }
        }
    }

    /**
     * {@inheritdoc}
     */
    public function disconnect()
    {
        if ($this->isConnected()) {
            phpiredis_reader_reset($this->reader);

The function phpiredis_reader_reset was not found. Maybe you did not declare it correctly or list all dependencies?

            socket_close($this->getResource());

            parent::disconnect();
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function write($buffer)
    {
        $socket = $this->getResource();

        while (($length = strlen($buffer)) > 0) {
            $written = socket_write($socket, $buffer, $length);

            if ($length === $written) {
                return;
            }

            if ($written === false) {
                $this->onConnectionError('Error while writing bytes to the server.');
            }

            $buffer = substr($buffer, $written);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $socket = $this->getResource();
        $reader = $this->reader;

        while (PHPIREDIS_READER_STATE_INCOMPLETE === $state = phpiredis_reader_get_state($reader)) {

The constant Predis\Connection\PHPIREDIS_READER_STATE_INCOMPLETE was not found. Maybe you did not declare it correctly or list all dependencies?

The function phpiredis_reader_get_state was not found. Maybe you did not declare it correctly or list all dependencies?

            if (@socket_recv($socket, $buffer, 4096, 0) === false || $buffer === '' || $buffer === null) {
                $this->emitSocketError();
            }

            phpiredis_reader_feed($reader, $buffer);

The function phpiredis_reader_feed was not found. Maybe you did not declare it correctly or list all dependencies?

        }

        if ($state === PHPIREDIS_READER_STATE_COMPLETE) {

The constant Predis\Connection\PHPIREDIS_READER_STATE_COMPLETE was not found. Maybe you did not declare it correctly or list all dependencies?

            return phpiredis_reader_get_reply($reader);

The function phpiredis_reader_get_reply was not found. Maybe you did not declare it correctly or list all dependencies?

        } else {
            $this->onProtocolError(phpiredis_reader_get_error($reader));

The function phpiredis_reader_get_error was not found. Maybe you did not declare it correctly or list all dependencies?

            return;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $arguments = $command->getArguments();
        array_unshift($arguments, $command->getId());

        $this->write(phpiredis_format_command($arguments));

The function phpiredis_format_command was not found. Maybe you did not declare it correctly or list all dependencies?

    }

    /**
     * {@inheritdoc}
     */
    public function __wakeup()
    {
        $this->assertExtensions();
        $this->reader = $this->createReader();
    }
}