14450042814

Committed 14 Apr 2025 03:51PM UTC coverage: 92.662% (-0.09%) from 92.752%

Build # 14450042814

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1527 from predis/vv-readme-8.0-support

Run Details

7147 of 7713 relevant lines covered (92.66%)

111.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.42

/src/Connection/Replication/SentinelReplication.php

<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2025 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\Replication;

use InvalidArgumentException;
use Predis\Command\Command;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\CommunicationException;
use Predis\Connection\AbstractAggregateConnection;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface as ConnectionFactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Connection\Parameters;
use Predis\Connection\ParametersInterface;
use Predis\Replication\ReplicationStrategy;
use Predis\Replication\RoleException;
use Predis\Response\Error;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ServerException;

/**
 * @author Daniele Alessandri <suppakilla@gmail.com>
 * @author Ville Mattila <ville@eventio.fi>
 */
class SentinelReplication extends AbstractAggregateConnection implements ReplicationInterface
{
    /**
     * @var NodeConnectionInterface
     */
    protected $master;

    /**
     * @var NodeConnectionInterface[]
     */
    protected $slaves = [];

    /**
     * @var NodeConnectionInterface[]
     */
    protected $pool = [];

    /**
     * @var NodeConnectionInterface
     */
    protected $current;

    /**
     * @var string
     */
    protected $service;

    /**
     * @var ConnectionFactoryInterface
     */
    protected $connectionFactory;

    /**
     * @var ReplicationStrategy
     */
    protected $strategy;

    /**
     * @var NodeConnectionInterface[]
     */
    protected $sentinels = [];

    /**
     * @var int
     */
    protected $sentinelIndex = 0;

    /**
     * @var NodeConnectionInterface
     */
    protected $sentinelConnection;

    /**
     * @var float
     */
    protected $sentinelTimeout = 0.100;

    /**
     * Max number of automatic retries of commands upon server failure.
     *
     * -1 = unlimited retry attempts
     *  0 = no retry attempts (fails immediately)
     *  n = fail only after n retry attempts
     *
     * @var int
     */
    protected $retryLimit = 20;

    /**
     * Time to wait in milliseconds before fetching a new configuration from one
     * of the sentinel servers.
     *
     * @var int
     */
    protected $retryWait = 1000;

    /**
     * Flag for automatic fetching of available sentinels.
     *
     * @var bool
     */
    protected $updateSentinels = false;

    /**
     * @param string                     $service           Name of the service for autodiscovery.
     * @param array                      $sentinels         Sentinel servers connection parameters.
     * @param ConnectionFactoryInterface $connectionFactory Connection factory instance.
     * @param ReplicationStrategy|null   $strategy          Replication strategy instance.
     */
    public function __construct(
        $service,
        array $sentinels,
        ConnectionFactoryInterface $connectionFactory,
        ?ReplicationStrategy $strategy = null
    ) {
        $this->sentinels = $sentinels;
        $this->service = $service;
        $this->connectionFactory = $connectionFactory;
        $this->strategy = $strategy ?: new ReplicationStrategy();
    }

    /**
     * Sets a default timeout for connections to sentinels.
     *
     * When "timeout" is present in the connection parameters of sentinels, its
     * value overrides the default sentinel timeout.
     *
     * @param float $timeout Timeout value.
     */
    public function setSentinelTimeout($timeout)
    {
        $this->sentinelTimeout = (float) $timeout;
    }

    /**
     * Sets the maximum number of retries for commands upon server failure.
     *
     * -1 = unlimited retry attempts
     *  0 = no retry attempts (fails immediately)
     *  n = fail only after n retry attempts
     *
     * @param int $retry Number of retry attempts.
     */
    public function setRetryLimit($retry)
    {
        $this->retryLimit = (int) $retry;
    }

    /**
     * Sets the time to wait (in milliseconds) before fetching a new configuration
     * from one of the sentinels.
     *
     * @param float $milliseconds Time to wait before the next attempt.
     */
    public function setRetryWait($milliseconds)
    {
        $this->retryWait = (float) $milliseconds;
    }

    /**
     * Set automatic fetching of available sentinels.
     *
     * @param bool $update Enable or disable automatic updates.
     */
    public function setUpdateSentinels($update)
    {
        $this->updateSentinels = (bool) $update;
    }

    /**
     * Resets the current connection.
     */
    protected function reset()
    {
        $this->current = null;
    }

    /**
     * Wipes the current list of master and slaves nodes.
     */
    protected function wipeServerList()
    {
        $this->reset();

        $this->master = null;
        $this->slaves = [];
        $this->pool = [];
    }

    /**
     * {@inheritdoc}
     */
    public function add(NodeConnectionInterface $connection)
    {
        $parameters = $connection->getParameters();
        $role = $parameters->role;

        if ('master' === $role) {
            $this->master = $connection;
        } elseif ('sentinel' === $role) {
            $this->sentinels[] = $connection;

            // sentinels are not considered part of the pool.
            return;
        } else {
            // everything else is considered a slave.
            $this->slaves[] = $connection;
        }

        $this->pool[(string) $connection] = $connection;

        $this->reset();
    }

    /**
     * {@inheritdoc}
     */
    public function remove(NodeConnectionInterface $connection)
    {
        if ($connection === $this->master) {
            $this->master = null;
        } elseif (false !== $id = array_search($connection, $this->slaves, true)) {
            unset($this->slaves[$id]);
        } elseif (false !== $id = array_search($connection, $this->sentinels, true)) {
            unset($this->sentinels[$id]);

            return true;
        } else {
            return false;
        }

        unset($this->pool[(string) $connection]);

        $this->reset();

        return true;
    }

    /**
     * Creates a new connection to a sentinel server.
     *
     * @return NodeConnectionInterface
     */
    protected function createSentinelConnection($parameters)
    {
        if ($parameters instanceof NodeConnectionInterface) {
            return $parameters;
        }

        if (is_string($parameters)) {
            $parameters = Parameters::parse($parameters);
        }

        if (is_array($parameters)) {
            // NOTE: sentinels do not accept SELECT command so we must
            // explicitly set it to NULL to avoid problems when using default
            // parameters set via client options.
            $parameters['database'] = null;

            // don't leak password from between configurations
            // https://github.com/predis/predis/pull/807/#discussion_r985764770
            if (!isset($parameters['password'])) {
                $parameters['password'] = null;
            }

            if (!isset($parameters['timeout'])) {
                $parameters['timeout'] = $this->sentinelTimeout;
            }
        }

        return $this->connectionFactory->create($parameters);
    }

    /**
     * Returns the current sentinel connection.
     *
     * If there is no active sentinel connection, a new connection is created.
     *
     * @return NodeConnectionInterface
     */
    public function getSentinelConnection()
    {
        if (!$this->sentinelConnection) {
            if ($this->sentinelIndex >= count($this->sentinels)) {
                $this->sentinelIndex = 0;
                throw new \Predis\ClientException('No sentinel server available for autodiscovery.');
            }

            $sentinel = $this->sentinels[$this->sentinelIndex];
            ++$this->sentinelIndex;
            $this->sentinelConnection = $this->createSentinelConnection($sentinel);
        }

        return $this->sentinelConnection;
    }

    /**
     * Fetches an updated list of sentinels from a sentinel.
     */
    public function updateSentinels()
    {
        SENTINEL_QUERY: {
            $sentinel = $this->getSentinelConnection();

            try {
                $payload = $sentinel->executeCommand(
                    RawCommand::create('SENTINEL', 'sentinels', $this->service)
                );

                $this->sentinels = [];
                $this->sentinelIndex = 0;
                // NOTE: sentinel server does not return itself, so we add it back.
                $this->sentinels[] = $sentinel->getParameters()->toArray();

                foreach ($payload as $sentinel) {
                    $this->sentinels[] = [
                        'host' => $sentinel[3],
                        'port' => $sentinel[5],
                        'role' => 'sentinel',
                    ];
                }
            } catch (ConnectionException $exception) {
                $this->sentinelConnection = null;

                goto SENTINEL_QUERY;
            }
        }
    }

    /**
     * Fetches the details for the master and slave servers from a sentinel.
     */
    public function querySentinel()
    {
        $this->wipeServerList();

        $this->updateSentinels();
        $this->getMaster();
        $this->getSlaves();
    }

    /**
     * Handles error responses returned by redis-sentinel.
     *
     * @param NodeConnectionInterface $sentinel Connection to a sentinel server.
     * @param ErrorResponseInterface  $error    Error response.
     */
    private function handleSentinelErrorResponse(NodeConnectionInterface $sentinel, ErrorResponseInterface $error)
    {
        if ($error->getErrorType() === 'IDONTKNOW') {
            throw new ConnectionException($sentinel, $error->getMessage());
        } else {
            throw new ServerException($error->getMessage());
        }
    }

    /**
     * Fetches the details for the master server from a sentinel.
     *
     * @param NodeConnectionInterface $sentinel Connection to a sentinel server.
     * @param string                  $service  Name of the service.
     *
     * @return array
     */
    protected function querySentinelForMaster(NodeConnectionInterface $sentinel, $service)
    {
        $payload = $sentinel->executeCommand(
            RawCommand::create('SENTINEL', 'get-master-addr-by-name', $service)
        );

        if ($payload === null) {
            throw new ServerException('ERR No such master with that name');
        }

        if ($payload instanceof ErrorResponseInterface) {
            $this->handleSentinelErrorResponse($sentinel, $payload);
        }

        return [
            'host' => $payload[0],
            'port' => $payload[1],
            'role' => 'master',
        ];
    }

    /**
     * Fetches the details for the slave servers from a sentinel.
     *
     * @param NodeConnectionInterface $sentinel Connection to a sentinel server.
     * @param string                  $service  Name of the service.
     *
     * @return array
     */
    protected function querySentinelForSlaves(NodeConnectionInterface $sentinel, $service)
    {
        $slaves = [];

        $payload = $sentinel->executeCommand(
            RawCommand::create('SENTINEL', 'slaves', $service)
        );

        if ($payload instanceof ErrorResponseInterface) {
            $this->handleSentinelErrorResponse($sentinel, $payload);
        }

        foreach ($payload as $slave) {
            $flags = explode(',', $slave[9]);

            if (array_intersect($flags, ['s_down', 'o_down', 'disconnected'])) {
                continue;
            }

            // ensure `master-link-status` is ok
            if (isset($slave[31]) && $slave[31] === 'err') {
                continue;
            }

            $slaves[] = [
                'host' => $slave[3],
                'port' => $slave[5],
                'role' => 'slave',
            ];
        }

        return $slaves;
    }

    /**
     * {@inheritdoc}
     */
    public function getCurrent()
    {
        return $this->current;
    }

    /**
     * {@inheritdoc}
     */
    public function getMaster()
    {
        if ($this->master) {
            return $this->master;
        }

        if ($this->updateSentinels) {
            $this->updateSentinels();
        }

        SENTINEL_QUERY: {
            $sentinel = $this->getSentinelConnection();

            try {
                $masterParameters = $this->querySentinelForMaster($sentinel, $this->service);
                $masterConnection = $this->connectionFactory->create($masterParameters);

                $this->add($masterConnection);
            } catch (ConnectionException $exception) {
                $this->sentinelConnection = null;

                goto SENTINEL_QUERY;
            }
        }

        return $masterConnection;
    }

    /**
     * {@inheritdoc}
     */
    public function getSlaves()
    {
        if ($this->slaves) {
            return array_values($this->slaves);
        }

        if ($this->updateSentinels) {
            $this->updateSentinels();
        }

        SENTINEL_QUERY: {
            $sentinel = $this->getSentinelConnection();

            try {
                $slavesParameters = $this->querySentinelForSlaves($sentinel, $this->service);

                foreach ($slavesParameters as $slaveParameters) {
                    $this->add($this->connectionFactory->create($slaveParameters));
                }
            } catch (ConnectionException $exception) {
                $this->sentinelConnection = null;

                goto SENTINEL_QUERY;
            }
        }

        return array_values($this->slaves);
    }

    /**
     * Returns a random slave.
     *
     * @return NodeConnectionInterface|null
     */
    protected function pickSlave()
    {
        $slaves = $this->getSlaves();

        return $slaves
            ? $slaves[rand(1, count($slaves)) - 1]
            : null;
    }

    /**
     * Returns the connection instance in charge for the given command.
     *
     * @param CommandInterface $command Command instance.
     *
     * @return NodeConnectionInterface
     */
    private function getConnectionInternal(CommandInterface $command)
    {
        if (!$this->current) {
            if ($this->strategy->isReadOperation($command) && $slave = $this->pickSlave()) {
                $this->current = $slave;
            } else {
                $this->current = $this->getMaster();
            }

            return $this->current;
        }

        if ($this->current === $this->master) {
            return $this->current;
        }

        if (!$this->strategy->isReadOperation($command)) {
            $this->current = $this->getMaster();
        }

        return $this->current;
    }

    /**
     * Asserts that the specified connection matches an expected role.
     *
     * @param NodeConnectionInterface $connection Connection to a redis server.
     * @param string                  $role       Expected role of the server ("master", "slave" or "sentinel").
     *
     * @throws RoleException|ConnectionException
     */
    protected function assertConnectionRole(NodeConnectionInterface $connection, $role)
    {
        $role = strtolower($role);
        $actualRole = $connection->executeCommand(RawCommand::create('ROLE'));

        if ($actualRole instanceof Error) {
            throw new ConnectionException($connection, $actualRole->getMessage());
        }

        if ($role !== $actualRole[0]) {
            throw new RoleException($connection, "Expected $role but got $actualRole[0] [$connection]");
        }
    }

    /**
     * {@inheritdoc}
     */
    public function getConnectionByCommand(CommandInterface $command)
    {
        $connection = $this->getConnectionInternal($command);

        if (!$connection->isConnected()) {
            // When we do not have any available slave in the pool we can expect
            // read-only operations to hit the master server.
            $expectedRole = $this->strategy->isReadOperation($command) && $this->slaves ? 'slave' : 'master';
            $this->assertConnectionRole($connection, $expectedRole);
        }

        return $connection;
    }

    /**
     * {@inheritdoc}
     */
    public function getConnectionById($id)
    {
        return $this->pool[$id] ?? null;
    }

    /**
     * Returns a connection by its role.
     *
     * @param string $role Connection role (`master`, `slave` or `sentinel`)
     *
     * @return NodeConnectionInterface|null
     */
    public function getConnectionByRole($role)
    {
        if ($role === 'master') {
            return $this->getMaster();
        } elseif ($role === 'slave') {
            return $this->pickSlave();
        } elseif ($role === 'sentinel') {
            return $this->getSentinelConnection();
        } else {
            return null;
        }
    }

    /**
     * Switches the internal connection in use by the backend.
     *
     * Sentinel connections are not considered as part of the pool, meaning that
     * trying to switch to a sentinel will throw an exception.
     *
     * @param NodeConnectionInterface $connection Connection instance in the pool.
     */
    public function switchTo(NodeConnectionInterface $connection)
    {
        if ($connection && $connection === $this->current) {
            return;
        }

        if ($connection !== $this->master && !in_array($connection, $this->slaves, true)) {
            throw new InvalidArgumentException('Invalid connection or connection not found.');
        }

        $connection->connect();

        if ($this->current) {
            $this->current->disconnect();
        }

        $this->current = $connection;
    }

    /**
     * {@inheritdoc}
     */
    public function switchToMaster()
    {
        $connection = $this->getConnectionByRole('master');
        $this->switchTo($connection);
    }

    /**
     * {@inheritdoc}
     */
    public function switchToSlave()
    {
        $connection = $this->getConnectionByRole('slave');
        $this->switchTo($connection);
    }

    /**
     * {@inheritdoc}
     */
    public function isConnected()
    {
        return $this->current ? $this->current->isConnected() : false;
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        if (!$this->current) {
            if (!$this->current = $this->pickSlave()) {
                $this->current = $this->getMaster();
            }
        }

        $this->current->connect();
    }

    /**
     * {@inheritdoc}
     */
    public function disconnect()
    {
        foreach ($this->pool as $connection) {
            $connection->disconnect();
        }
    }

    /**
     * Retries the execution of a command upon server failure after asking a new
     * configuration to one of the sentinels.
     *
     * @param CommandInterface $command Command instance.
     * @param string           $method  Actual method.
     *
     * @return mixed
     */
    private function retryCommandOnFailure(CommandInterface $command, $method)
    {
        $retries = 0;

        while ($retries <= $this->retryLimit) {
            try {
                $response = $this->getConnectionByCommand($command)->$method($command);
                break;
            } catch (CommunicationException $exception) {
                $this->wipeServerList();
                $exception->getConnection()->disconnect();

                if ($retries === $this->retryLimit) {
                    throw $exception;
                }

                usleep($this->retryWait * 1000);

                ++$retries;
            }
        }

        return $response;
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $this->retryCommandOnFailure($command, __FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function readResponse(CommandInterface $command)
    {
        return $this->retryCommandOnFailure($command, __FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function executeCommand(CommandInterface $command)
    {
        return $this->retryCommandOnFailure($command, __FUNCTION__);
    }

    /**
     * Returns the underlying replication strategy.
     *
     * @return ReplicationStrategy
     */
    public function getReplicationStrategy()
    {
        return $this->strategy;
    }

    /**
     * {@inheritdoc}
     */
    public function __sleep()
    {
        return [
            'master', 'slaves', 'pool', 'service', 'sentinels', 'connectionFactory', 'strategy',
        ];
    }

    /**
     * {@inheritdoc}
     */
    public function getParameters(): ?ParametersInterface
    {
        if (isset($this->master)) {
            return $this->master->getParameters();
        }

        if (!empty($this->slaves)) {
            return $this->slaves[0]->getParameters();
        }

        if (!empty($this->sentinels)) {
            return $this->sentinels[0]->getParameters();
        }

        return null;
    }
}

1	<?php
2
3	/*
4	* This file is part of the Predis package.
5	*
6	* (c) 2009-2020 Daniele Alessandri
7	* (c) 2021-2025 Till Krüss
8	*
9	* For the full copyright and license information, please view the LICENSE
10	* file that was distributed with this source code.
11	*/
12
13	namespace Predis\Connection\Replication;
14
15	use InvalidArgumentException;
16	use Predis\Command\Command;
17	use Predis\Command\CommandInterface;
18	use Predis\Command\RawCommand;
19	use Predis\CommunicationException;
20	use Predis\Connection\AbstractAggregateConnection;
21	use Predis\Connection\ConnectionException;
22	use Predis\Connection\FactoryInterface as ConnectionFactoryInterface;
23	use Predis\Connection\NodeConnectionInterface;
24	use Predis\Connection\Parameters;
25	use Predis\Connection\ParametersInterface;
26	use Predis\Replication\ReplicationStrategy;
27	use Predis\Replication\RoleException;
28	use Predis\Response\Error;
29	use Predis\Response\ErrorInterface as ErrorResponseInterface;
30	use Predis\Response\ServerException;
31
32	/**
33	* @author Daniele Alessandri <suppakilla@gmail.com>
34	* @author Ville Mattila <ville@eventio.fi>
35	*/
36	class SentinelReplication extends AbstractAggregateConnection implements ReplicationInterface
37	{
38	/**
39	* @var NodeConnectionInterface
40	*/
41	protected $master;
42
43	/**
44	* @var NodeConnectionInterface[]
45	*/
46	protected $slaves = [];
47
48	/**
49	* @var NodeConnectionInterface[]
50	*/
51	protected $pool = [];
52
53	/**
54	* @var NodeConnectionInterface
55	*/
56	protected $current;
57
58	/**
59	* @var string
60	*/
61	protected $service;
62
63	/**
64	* @var ConnectionFactoryInterface
65	*/
66	protected $connectionFactory;
67
68	/**
69	* @var ReplicationStrategy
70	*/
71	protected $strategy;
72
73	/**
74	* @var NodeConnectionInterface[]
75	*/
76	protected $sentinels = [];
77
78	/**
79	* @var int
80	*/
81	protected $sentinelIndex = 0;
82
83	/**
84	* @var NodeConnectionInterface
85	*/
86	protected $sentinelConnection;
87
88	/**
89	* @var float
90	*/
91	protected $sentinelTimeout = 0.100;
92
93	/**
94	* Max number of automatic retries of commands upon server failure.
95	*
96	* -1 = unlimited retry attempts
97	* 0 = no retry attempts (fails immediately)
98	* n = fail only after n retry attempts
99	*
100	* @var int
101	*/
102	protected $retryLimit = 20;
103
104	/**
105	* Time to wait in milliseconds before fetching a new configuration from one
106	* of the sentinel servers.
107	*
108	* @var int
109	*/
110	protected $retryWait = 1000;
111
112	/**
113	* Flag for automatic fetching of available sentinels.
114	*
115	* @var bool
116	*/
117	protected $updateSentinels = false;
118
119	/**
120	* @param string $service Name of the service for autodiscovery.
121	* @param array $sentinels Sentinel servers connection parameters.
122	* @param ConnectionFactoryInterface $connectionFactory Connection factory instance.
123	* @param ReplicationStrategy\|null $strategy Replication strategy instance.
124	*/
125	public function __construct(	51✔
126	$service,
127	array $sentinels,
128	ConnectionFactoryInterface $connectionFactory,
129	?ReplicationStrategy $strategy = null
130	) {
131	$this->sentinels = $sentinels;	51✔
132	$this->service = $service;	51✔
133	$this->connectionFactory = $connectionFactory;	51✔
134	$this->strategy = $strategy ?: new ReplicationStrategy();	51✔
135	}
136
137	/**
138	* Sets a default timeout for connections to sentinels.
139	*
140	* When "timeout" is present in the connection parameters of sentinels, its
141	* value overrides the default sentinel timeout.
142	*
143	* @param float $timeout Timeout value.
144	*/
145	public function setSentinelTimeout($timeout)	×
146	{
147	$this->sentinelTimeout = (float) $timeout;	×
148	}
149
150	/**
151	* Sets the maximum number of retries for commands upon server failure.
152	*
153	* -1 = unlimited retry attempts
154	* 0 = no retry attempts (fails immediately)
155	* n = fail only after n retry attempts
156	*
157	* @param int $retry Number of retry attempts.
158	*/
159	public function setRetryLimit($retry)	×
160	{
161	$this->retryLimit = (int) $retry;	×
162	}
163
164	/**
165	* Sets the time to wait (in milliseconds) before fetching a new configuration
166	* from one of the sentinels.
167	*
168	* @param float $milliseconds Time to wait before the next attempt.
169	*/
170	public function setRetryWait($milliseconds)	47✔
171	{
172	$this->retryWait = (float) $milliseconds;	47✔
173	}
174
175	/**
176	* Set automatic fetching of available sentinels.
177	*
178	* @param bool $update Enable or disable automatic updates.
179	*/
180	public function setUpdateSentinels($update)	×
181	{
182	$this->updateSentinels = (bool) $update;	×
183	}
184
185	/**
186	* Resets the current connection.
187	*/
188	protected function reset()	32✔
189	{
190	$this->current = null;	32✔
191	}
192
193	/**
194	* Wipes the current list of master and slaves nodes.
195	*/
196	protected function wipeServerList()	5✔
197	{
198	$this->reset();	5✔
199
200	$this->master = null;	5✔
201	$this->slaves = [];	5✔
202	$this->pool = [];	5✔
203	}
204
205	/**
206	* {@inheritdoc}
207	*/
208	public function add(NodeConnectionInterface $connection)	32✔
209	{
210	$parameters = $connection->getParameters();	32✔
211	$role = $parameters->role;	32✔
212
213	if ('master' === $role) {	32✔
214	$this->master = $connection;	29✔
215	} elseif ('sentinel' === $role) {	24✔
216	$this->sentinels[] = $connection;	2✔
217
218	// sentinels are not considered part of the pool.
219	return;	2✔
220	} else {
221	// everything else is considered a slave.
222	$this->slaves[] = $connection;	24✔
223	}
224
225	$this->pool[(string) $connection] = $connection;	32✔
226
227	$this->reset();	32✔
228	}
229
230	/**
231	* {@inheritdoc}
232	*/
233	public function remove(NodeConnectionInterface $connection)	1✔
234	{
235	if ($connection === $this->master) {	1✔
236	$this->master = null;	×
237	} elseif (false !== $id = array_search($connection, $this->slaves, true)) {	1✔
238	unset($this->slaves[$id]);	1✔
239	} elseif (false !== $id = array_search($connection, $this->sentinels, true)) {	1✔
240	unset($this->sentinels[$id]);	1✔
241
242	return true;	1✔
243	} else {
244	return false;	×
245	}
246
247	unset($this->pool[(string) $connection]);	1✔
248
249	$this->reset();	1✔
250
251	return true;	1✔
252	}
253
254	/**
255	* Creates a new connection to a sentinel server.
256	*
257	* @return NodeConnectionInterface
258	*/
259	protected function createSentinelConnection($parameters)	26✔
260	{
261	if ($parameters instanceof NodeConnectionInterface) {	26✔
262	return $parameters;	20✔
263	}
264
265	if (is_string($parameters)) {	6✔
266	$parameters = Parameters::parse($parameters);	4✔
267	}
268
269	if (is_array($parameters)) {	6✔
270	// NOTE: sentinels do not accept SELECT command so we must
271	// explicitly set it to NULL to avoid problems when using default
272	// parameters set via client options.
273	$parameters['database'] = null;	4✔
274
275	// don't leak password from between configurations
276	// https://github.com/predis/predis/pull/807/#discussion_r985764770
277	if (!isset($parameters['password'])) {	4✔
278	$parameters['password'] = null;	2✔
279	}
280
281	if (!isset($parameters['timeout'])) {	4✔
282	$parameters['timeout'] = $this->sentinelTimeout;	3✔
283	}
284	}
285
286	return $this->connectionFactory->create($parameters);	6✔
287	}
288
289	/**
290	* Returns the current sentinel connection.
291	*
292	* If there is no active sentinel connection, a new connection is created.
293	*
294	* @return NodeConnectionInterface
295	*/
296	public function getSentinelConnection()	28✔
297	{
298	if (!$this->sentinelConnection) {	28✔
299	if ($this->sentinelIndex >= count($this->sentinels)) {	28✔
300	$this->sentinelIndex = 0;	7✔
301	throw new \Predis\ClientException('No sentinel server available for autodiscovery.');	7✔
302	}
303
304	$sentinel = $this->sentinels[$this->sentinelIndex];	26✔
305	++$this->sentinelIndex;	26✔
306	$this->sentinelConnection = $this->createSentinelConnection($sentinel);	26✔
307	}
308
309	return $this->sentinelConnection;	26✔
310	}
311
312	/**
313	* Fetches an updated list of sentinels from a sentinel.
314	*/
315	public function updateSentinels()	5✔
316	{
317	SENTINEL_QUERY: {	5✔
318	$sentinel = $this->getSentinelConnection();	5✔
319
320	try {	5✔
321	$payload = $sentinel->executeCommand(	5✔
322	RawCommand::create('SENTINEL', 'sentinels', $this->service)	5✔
323	);	5✔
324
325	$this->sentinels = [];	5✔
326	$this->sentinelIndex = 0;	5✔
327	// NOTE: sentinel server does not return itself, so we add it back.
328	$this->sentinels[] = $sentinel->getParameters()->toArray();	5✔
329
330	foreach ($payload as $sentinel) {	4✔
331	$this->sentinels[] = [	5✔
332	'host' => $sentinel[3],	5✔
333	'port' => $sentinel[5],	5✔
334	'role' => 'sentinel',	5✔
335	];	5✔
336	}	5✔
337	} catch (ConnectionException $exception) {	3✔
338	$this->sentinelConnection = null;	5✔
339
340	goto SENTINEL_QUERY;	5✔
341	}	5✔
342	}	5✔
343	}
344
345	/**
346	* Fetches the details for the master and slave servers from a sentinel.
347	*/
348	public function querySentinel()	1✔
349	{
350	$this->wipeServerList();	1✔
351
352	$this->updateSentinels();	1✔
353	$this->getMaster();	1✔
354	$this->getSlaves();	1✔
355	}
356
357	/**
358	* Handles error responses returned by redis-sentinel.
359	*
360	* @param NodeConnectionInterface $sentinel Connection to a sentinel server.
361	* @param ErrorResponseInterface $error Error response.
362	*/
363	private function handleSentinelErrorResponse(NodeConnectionInterface $sentinel, ErrorResponseInterface $error)	×
364	{
365	if ($error->getErrorType() === 'IDONTKNOW') {	×
366	throw new ConnectionException($sentinel, $error->getMessage());	×
367	} else {
368	throw new ServerException($error->getMessage());	×
369	}
370	}
371
372	/**
373	* Fetches the details for the master server from a sentinel.
374	*
375	* @param NodeConnectionInterface $sentinel Connection to a sentinel server.
376	* @param string $service Name of the service.
377	*
378	* @return array
379	*/
380	protected function querySentinelForMaster(NodeConnectionInterface $sentinel, $service)	8✔
381	{
382	$payload = $sentinel->executeCommand(	8✔
383	RawCommand::create('SENTINEL', 'get-master-addr-by-name', $service)	8✔
384	);	8✔
385
386	if ($payload === null) {	6✔
387	throw new ServerException('ERR No such master with that name');	1✔
388	}
389
390	if ($payload instanceof ErrorResponseInterface) {	5✔
391	$this->handleSentinelErrorResponse($sentinel, $payload);	×
392	}
393
394	return [	5✔
395	'host' => $payload[0],	5✔
396	'port' => $payload[1],	5✔
397	'role' => 'master',	5✔
398	];	5✔
399	}
400
401	/**
402	* Fetches the details for the slave servers from a sentinel.
403	*
404	* @param NodeConnectionInterface $sentinel Connection to a sentinel server.
405	* @param string $service Name of the service.
406	*
407	* @return array
408	*/
409	protected function querySentinelForSlaves(NodeConnectionInterface $sentinel, $service)	8✔
410	{
411	$slaves = [];	8✔
412
413	$payload = $sentinel->executeCommand(	8✔
414	RawCommand::create('SENTINEL', 'slaves', $service)	8✔
415	);	8✔
416
417	if ($payload instanceof ErrorResponseInterface) {	7✔
418	$this->handleSentinelErrorResponse($sentinel, $payload);	×
419	}
420
421	foreach ($payload as $slave) {	7✔
422	$flags = explode(',', $slave[9]);	6✔
423
424	if (array_intersect($flags, ['s_down', 'o_down', 'disconnected'])) {	6✔
425	continue;	1✔
426	}
427
428	// ensure `master-link-status` is ok
429	if (isset($slave[31]) && $slave[31] === 'err') {	5✔
430	continue;	×
431	}
432
433	$slaves[] = [	5✔
434	'host' => $slave[3],	5✔
435	'port' => $slave[5],	5✔
436	'role' => 'slave',	5✔
437	];	5✔
438	}
439
440	return $slaves;	7✔
441	}
442
443	/**
444	* {@inheritdoc}
445	*/
446	public function getCurrent()	3✔
447	{
448	return $this->current;	3✔
449	}
450
451	/**
452	* {@inheritdoc}
453	*/
454	public function getMaster()	18✔
455	{
456	if ($this->master) {	18✔
457	return $this->master;	14✔
458	}
459
460	if ($this->updateSentinels) {	8✔
461	$this->updateSentinels();	×
462	}
463
464	SENTINEL_QUERY: {	8✔
465	$sentinel = $this->getSentinelConnection();	8✔
466
467	try {	8✔
468	$masterParameters = $this->querySentinelForMaster($sentinel, $this->service);	8✔
469	$masterConnection = $this->connectionFactory->create($masterParameters);	8✔
470
471	$this->add($masterConnection);	8✔
472	} catch (ConnectionException $exception) {	3✔
473	$this->sentinelConnection = null;	8✔
474
475	goto SENTINEL_QUERY;	8✔
476	}	8✔
477	}	8✔
478
479	return $masterConnection;	5✔
480	}
481
482	/**
483	* {@inheritdoc}
484	*/
485	public function getSlaves()	18✔
486	{
487	if ($this->slaves) {	18✔
488	return array_values($this->slaves);	11✔
489	}
490
491	if ($this->updateSentinels) {	9✔
492	$this->updateSentinels();	×
493	}
494
495	SENTINEL_QUERY: {	9✔
496	$sentinel = $this->getSentinelConnection();	9✔
497
498	try {	9✔
499	$slavesParameters = $this->querySentinelForSlaves($sentinel, $this->service);	9✔
500
501	foreach ($slavesParameters as $slaveParameters) {	7✔
502	$this->add($this->connectionFactory->create($slaveParameters));	9✔
503	}	9✔
504	} catch (ConnectionException $exception) {	1✔
505	$this->sentinelConnection = null;	9✔
506
507	goto SENTINEL_QUERY;	9✔
508	}	9✔
509	}	9✔
510
511	return array_values($this->slaves);	7✔
512	}
513
514	/**
515	* Returns a random slave.
516	*
517	* @return NodeConnectionInterface\|null
518	*/
519	protected function pickSlave()	13✔
520	{
521	$slaves = $this->getSlaves();	13✔
522
523	return $slaves	12✔
524	? $slaves[rand(1, count($slaves)) - 1]	10✔
525	: null;	12✔
526	}
527
528	/**
529	* Returns the connection instance in charge for the given command.
530	*
531	* @param CommandInterface $command Command instance.
532	*
533	* @return NodeConnectionInterface
534	*/
535	private function getConnectionInternal(CommandInterface $command)	11✔
536	{
537	if (!$this->current) {	11✔
538	if ($this->strategy->isReadOperation($command) && $slave = $this->pickSlave()) {	11✔
539	$this->current = $slave;	4✔
540	} else {
541	$this->current = $this->getMaster();	7✔
542	}
543
544	return $this->current;	11✔
545	}
546
547	if ($this->current === $this->master) {	5✔
548	return $this->current;	3✔
549	}
550
551	if (!$this->strategy->isReadOperation($command)) {	3✔
552	$this->current = $this->getMaster();	2✔
553	}
554
555	return $this->current;	3✔
556	}
557
558	/**
559	* Asserts that the specified connection matches an expected role.
560	*
561	* @param NodeConnectionInterface $connection Connection to a redis server.
562	* @param string $role Expected role of the server ("master", "slave" or "sentinel").
563	*
564	* @throws RoleException\|ConnectionException
565	*/
566	protected function assertConnectionRole(NodeConnectionInterface $connection, $role)	5✔
567	{
568	$role = strtolower($role);	5✔
569	$actualRole = $connection->executeCommand(RawCommand::create('ROLE'));	5✔
570
571	if ($actualRole instanceof Error) {	5✔
572	throw new ConnectionException($connection, $actualRole->getMessage());	×
573	}
574
575	if ($role !== $actualRole[0]) {	5✔
576	throw new RoleException($connection, "Expected $role but got $actualRole[0] [$connection]");	1✔
577	}
578	}
579
580	/**
581	* {@inheritdoc}
582	*/
583	public function getConnectionByCommand(CommandInterface $command)	11✔
584	{
585	$connection = $this->getConnectionInternal($command);	11✔
586
587	if (!$connection->isConnected()) {	11✔
588	// When we do not have any available slave in the pool we can expect
589	// read-only operations to hit the master server.
590	$expectedRole = $this->strategy->isReadOperation($command) && $this->slaves ? 'slave' : 'master';	5✔
591	$this->assertConnectionRole($connection, $expectedRole);	5✔
592	}
593
594	return $connection;	10✔
595	}
596
597	/**
598	* {@inheritdoc}
599	*/
600	public function getConnectionById($id)	5✔
601	{
602	return $this->pool[$id] ?? null;	5✔
603	}
604
605	/**
606	* Returns a connection by its role.
607	*
608	* @param string $role Connection role (`master`, `slave` or `sentinel`)
609	*
610	* @return NodeConnectionInterface\|null
611	*/
612	public function getConnectionByRole($role)	5✔
613	{
614	if ($role === 'master') {	5✔
615	return $this->getMaster();	3✔
616	} elseif ($role === 'slave') {	4✔
617	return $this->pickSlave();	3✔
618	} elseif ($role === 'sentinel') {	3✔
619	return $this->getSentinelConnection();	2✔
620	} else {
621	return null;	1✔
622	}
623	}
624
625	/**
626	* Switches the internal connection in use by the backend.
627	*
628	* Sentinel connections are not considered as part of the pool, meaning that
629	* trying to switch to a sentinel will throw an exception.
630	*
631	* @param NodeConnectionInterface $connection Connection instance in the pool.
632	*/
633	public function switchTo(NodeConnectionInterface $connection)	4✔
634	{
635	if ($connection && $connection === $this->current) {	4✔
636	return;	×
637	}
638
639	if ($connection !== $this->master && !in_array($connection, $this->slaves, true)) {	4✔
640	throw new InvalidArgumentException('Invalid connection or connection not found.');	1✔
641	}
642
643	$connection->connect();	3✔
644
645	if ($this->current) {	3✔
646	$this->current->disconnect();	1✔
647	}
648
649	$this->current = $connection;	3✔
650	}
651
652	/**
653	* {@inheritdoc}
654	*/
655	public function switchToMaster()	1✔
656	{
657	$connection = $this->getConnectionByRole('master');	1✔
658	$this->switchTo($connection);	1✔
659	}
660
661	/**
662	* {@inheritdoc}
663	*/
664	public function switchToSlave()	1✔
665	{
666	$connection = $this->getConnectionByRole('slave');	1✔
667	$this->switchTo($connection);	1✔
668	}
669
670	/**
671	* {@inheritdoc}
672	*/
673	public function isConnected()	1✔
674	{
675	return $this->current ? $this->current->isConnected() : false;	1✔
676	}
677
678	/**
679	* {@inheritdoc}
680	*/
681	public function connect()	5✔
682	{
683	if (!$this->current) {	5✔
684	if (!$this->current = $this->pickSlave()) {	5✔
685	$this->current = $this->getMaster();	1✔
686	}
687	}
688
689	$this->current->connect();	4✔
690	}
691
692	/**
693	* {@inheritdoc}
694	*/
695	public function disconnect()	1✔
696	{
697	foreach ($this->pool as $connection) {	1✔
698	$connection->disconnect();	1✔
699	}
700	}
701
702	/**
703	* Retries the execution of a command upon server failure after asking a new
704	* configuration to one of the sentinels.
705	*
706	* @param CommandInterface $command Command instance.
707	* @param string $method Actual method.
708	*
709	* @return mixed
710	*/
711	private function retryCommandOnFailure(CommandInterface $command, $method)	5✔
712	{
713	$retries = 0;	5✔
714
715	while ($retries <= $this->retryLimit) {	5✔
716	try {
717	$response = $this->getConnectionByCommand($command)->$method($command);	5✔
718	break;	3✔
719	} catch (CommunicationException $exception) {	4✔
720	$this->wipeServerList();	4✔
721	$exception->getConnection()->disconnect();	4✔
722
723	if ($retries === $this->retryLimit) {	4✔
724	throw $exception;	×
725	}
726
727	usleep($this->retryWait * 1000);	4✔
728
729	++$retries;	4✔
730	}
731	}
732
733	return $response;	3✔
734	}
735
736	/**
737	* {@inheritdoc}
738	*/
739	public function writeRequest(CommandInterface $command)	×
740	{
741	$this->retryCommandOnFailure($command, __FUNCTION__);	×
742	}
743
744	/**
745	* {@inheritdoc}
746	*/
747	public function readResponse(CommandInterface $command)	×
748	{
749	return $this->retryCommandOnFailure($command, __FUNCTION__);	×
750	}
751
752	/**
753	* {@inheritdoc}
754	*/
755	public function executeCommand(CommandInterface $command)	5✔
756	{
757	return $this->retryCommandOnFailure($command, __FUNCTION__);	5✔
758	}
759
760	/**
761	* Returns the underlying replication strategy.
762	*
763	* @return ReplicationStrategy
764	*/
765	public function getReplicationStrategy()	2✔
766	{
767	return $this->strategy;	2✔
768	}
769
770	/**
771	* {@inheritdoc}
772	*/
773	public function __sleep()	1✔
774	{
775	return [	1✔
776	'master', 'slaves', 'pool', 'service', 'sentinels', 'connectionFactory', 'strategy',	1✔
777	];	1✔
778	}
779
780	/**
781	* {@inheritdoc}
782	*/
783	public function getParameters(): ?ParametersInterface	3✔
784	{
785	if (isset($this->master)) {	3✔
786	return $this->master->getParameters();	1✔
787	}
788
789	if (!empty($this->slaves)) {	2✔
790	return $this->slaves[0]->getParameters();	1✔
791	}
792
793	if (!empty($this->sentinels)) {	1✔
794	return $this->sentinels[0]->getParameters();	1✔
795	}
796
797	return null;	×
798	}
799	}

predis / predis / 14450042814

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous