Dereku/snappymail
1
0
Fork 0
mirror of https://github.com/the-djmaze/snappymail.git synced 2026-08-29 20:19:22 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Predis to v2.2.2

Browse source
This commit is contained in:
the-djmaze 2024-04-08 16:29:55 +02:00
parent 9510347d3a
commit 969dca5f7e
449 changed files with 22013 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

23
plugins/cache-redis/Predis/Connection/Cluster/ClusterInterface.php Normal file
View file

@ -0,0 +1,23 @@
<?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\Cluster;
use Predis\Connection\AggregateConnectionInterface;
/**
* Defines a cluster of Redis servers formed by aggregating multiple connection
* instances to single Redis nodes.
*/
interface ClusterInterface extends AggregateConnectionInterface
{
}

244
plugins/cache-redis/Predis/Connection/Cluster/PredisCluster.php Normal file
View file

@ -0,0 +1,244 @@
<?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\Cluster;
use ArrayIterator;
use Countable;
use IteratorAggregate;
use Predis\Cluster\PredisStrategy;
use Predis\Cluster\StrategyInterface;
use Predis\Command\CommandInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\NotSupportedException;
use ReturnTypeWillChange;
use Traversable;
/**
* Abstraction for a cluster of aggregate connections to various Redis servers
* implementing client-side sharding based on pluggable distribution strategies.
*/
class PredisCluster implements ClusterInterface, IteratorAggregate, Countable
{
/**
* @var NodeConnectionInterface[]
*/
private $pool = [];
/**
* @var NodeConnectionInterface[]
*/
private $aliases = [];
/**
* @var StrategyInterface
*/
private $strategy;
/**
* @var \Predis\Cluster\Distributor\DistributorInterface
*/
private $distributor;
/**
* @param StrategyInterface $strategy Optional cluster strategy.
*/
public function __construct(StrategyInterface $strategy = null)
{
$this->strategy = $strategy ?: new PredisStrategy();
$this->distributor = $this->strategy->getDistributor();
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
foreach ($this->pool as $connection) {
if ($connection->isConnected()) {
return true;
}
}
return false;
}
/**
* {@inheritdoc}
*/
public function connect()
{
foreach ($this->pool as $connection) {
$connection->connect();
}
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
foreach ($this->pool as $connection) {
$connection->disconnect();
}
}
/**
* {@inheritdoc}
*/
public function add(NodeConnectionInterface $connection)
{
$parameters = $connection->getParameters();
$this->pool[(string) $connection] = $connection;
if (isset($parameters->alias)) {
$this->aliases[$parameters->alias] = $connection;
}
$this->distributor->add($connection, $parameters->weight);
}
/**
* {@inheritdoc}
*/
public function remove(NodeConnectionInterface $connection)
{
if (false !== $id = array_search($connection, $this->pool, true)) {
unset($this->pool[$id]);
$this->distributor->remove($connection);
if ($this->aliases && $alias = $connection->getParameters()->alias) {
unset($this->aliases[$alias]);
}
return true;
}
return false;
}
/**
* {@inheritdoc}
*/
public function getConnectionByCommand(CommandInterface $command)
{
$slot = $this->strategy->getSlot($command);
if (!isset($slot)) {
throw new NotSupportedException(
"Cannot use '{$command->getId()}' over clusters of connections."
);
}
return $this->distributor->getBySlot($slot);
}
/**
* {@inheritdoc}
*/
public function getConnectionById($id)
{
return $this->pool[$id] ?? null;
}
/**
* Returns a connection instance by its alias.
*
* @param string $alias Connection alias.
*
* @return NodeConnectionInterface|null
*/
public function getConnectionByAlias($alias)
{
return $this->aliases[$alias] ?? null;
}
/**
* Retrieves a connection instance by slot.
*
* @param string $slot Slot name.
*
* @return NodeConnectionInterface|null
*/
public function getConnectionBySlot($slot)
{
return $this->distributor->getBySlot($slot);
}
/**
* Retrieves a connection instance from the cluster using a key.
*
* @param string $key Key string.
*
* @return NodeConnectionInterface
*/
public function getConnectionByKey($key)
{
$hash = $this->strategy->getSlotByKey($key);
return $this->distributor->getBySlot($hash);
}
/**
* Returns the underlying command hash strategy used to hash commands by
* using keys found in their arguments.
*
* @return StrategyInterface
*/
public function getClusterStrategy()
{
return $this->strategy;
}
/**
* @return int
*/
#[ReturnTypeWillChange]
public function count()
{
return count($this->pool);
}
/**
* @return Traversable<string, NodeConnectionInterface>
*/
#[ReturnTypeWillChange]
public function getIterator()
{
return new ArrayIterator($this->pool);
}
/**
* {@inheritdoc}
*/
public function writeRequest(CommandInterface $command)
{
$this->getConnectionByCommand($command)->writeRequest($command);
}
/**
* {@inheritdoc}
*/
public function readResponse(CommandInterface $command)
{
return $this->getConnectionByCommand($command)->readResponse($command);
}
/**
* {@inheritdoc}
*/
public function executeCommand(CommandInterface $command)
{
return $this->getConnectionByCommand($command)->executeCommand($command);
}
}

673
plugins/cache-redis/Predis/Connection/Cluster/RedisCluster.php Normal file
View file

@ -0,0 +1,673 @@
<?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\Cluster;
use ArrayIterator;
use Countable;
use IteratorAggregate;
use OutOfBoundsException;
use Predis\ClientException;
use Predis\Cluster\RedisStrategy as RedisClusterStrategy;
use Predis\Cluster\SlotMap;
use Predis\Cluster\StrategyInterface;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ServerException;
use ReturnTypeWillChange;
use Throwable;
use Traversable;
/**
* Abstraction for a Redis-backed cluster of nodes (Redis >= 3.0.0).
*
* This connection backend offers smart support for redis-cluster by handling
* automatic slots map (re)generation upon -MOVED or -ASK responses returned by
* Redis when redirecting a client to a different node.
*
* The cluster can be pre-initialized using only a subset of the actual nodes in
* the cluster, Predis will do the rest by adjusting the slots map and creating
* the missing underlying connection instances on the fly.
*
* It is possible to pre-associate connections to a slots range with the "slots"
* parameter in the form "$first-$last". This can greatly reduce runtime node
* guessing and redirections.
*
* It is also possible to ask for the full and updated slots map directly to one
* of the nodes and optionally enable such a behaviour upon -MOVED redirections.
* Asking for the cluster configuration to Redis is actually done by issuing a
* CLUSTER SLOTS command to a random node in the pool.
*/
class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
{
private $useClusterSlots = true;
private $pool = [];
private $slots = [];
private $slotmap;
private $strategy;
private $connections;
private $retryLimit = 5;
private $retryInterval = 10;
/**
* @param FactoryInterface $connections Optional connection factory.
* @param StrategyInterface $strategy Optional cluster strategy.
*/
public function __construct(
FactoryInterface $connections,
StrategyInterface $strategy = null
) {
$this->connections = $connections;
$this->strategy = $strategy ?: new RedisClusterStrategy();
$this->slotmap = new SlotMap();
}
/**
* 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 initial retry interval (milliseconds).
*
* @param int $retryInterval Milliseconds between retries.
*/
public function setRetryInterval($retryInterval)
{
$this->retryInterval = (int) $retryInterval;
}
/**
* Returns the retry interval (milliseconds).
*
* @return int Milliseconds between retries.
*/
public function getRetryInterval()
{
return (int) $this->retryInterval;
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
foreach ($this->pool as $connection) {
if ($connection->isConnected()) {
return true;
}
}
return false;
}
/**
* {@inheritdoc}
*/
public function connect()
{
if ($connection = $this->getRandomConnection()) {
$connection->connect();
}
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
foreach ($this->pool as $connection) {
$connection->disconnect();
}
}
/**
* {@inheritdoc}
*/
public function add(NodeConnectionInterface $connection)
{
$this->pool[(string) $connection] = $connection;
$this->slotmap->reset();
}
/**
* {@inheritdoc}
*/
public function remove(NodeConnectionInterface $connection)
{
if (false !== $id = array_search($connection, $this->pool, true)) {
$this->slotmap->reset();
$this->slots = array_diff($this->slots, [$connection]);
unset($this->pool[$id]);
return true;
}
return false;
}
/**
* Removes a connection instance by using its identifier.
*
* @param string $connectionID Connection identifier.
*
* @return bool True if the connection was in the pool.
*/
public function removeById($connectionID)
{
if (isset($this->pool[$connectionID])) {
$this->slotmap->reset();
$this->slots = array_diff($this->slots, [$connectionID]);
unset($this->pool[$connectionID]);
return true;
}
return false;
}
/**
* Generates the current slots map by guessing the cluster configuration out
* of the connection parameters of the connections in the pool.
*
* Generation is based on the same algorithm used by Redis to generate the
* cluster, so it is most effective when all of the connections supplied on
* initialization have the "slots" parameter properly set accordingly to the
* current cluster configuration.
*/
public function buildSlotMap()
{
$this->slotmap->reset();
foreach ($this->pool as $connectionID => $connection) {
$parameters = $connection->getParameters();
if (!isset($parameters->slots)) {
continue;
}
foreach (explode(',', $parameters->slots) as $slotRange) {
$slots = explode('-', $slotRange, 2);
if (!isset($slots[1])) {
$slots[1] = $slots[0];
}
$this->slotmap->setSlots($slots[0], $slots[1], $connectionID);
}
}
}
/**
* Queries the specified node of the cluster to fetch the updated slots map.
*
* When the connection fails, this method tries to execute the same command
* on a different connection picked at random from the pool of known nodes,
* up until the retry limit is reached.
*
* @param NodeConnectionInterface $connection Connection to a node of the cluster.
*
* @return mixed
*/
private function queryClusterNodeForSlotMap(NodeConnectionInterface $connection)
{
$retries = 0;
$retryAfter = $this->retryInterval;
$command = RawCommand::create('CLUSTER', 'SLOTS');
while ($retries <= $this->retryLimit) {
try {
$response = $connection->executeCommand($command);
break;
} catch (ConnectionException $exception) {
$connection = $exception->getConnection();
$connection->disconnect();
$this->remove($connection);
if ($retries === $this->retryLimit) {
throw $exception;
}
if (!$connection = $this->getRandomConnection()) {
throw new ClientException('No connections left in the pool for `CLUSTER SLOTS`');
}
usleep($retryAfter * 1000);
$retryAfter = $retryAfter * 2;
++$retries;
}
}
return $response;
}
/**
* Generates an updated slots map fetching the cluster configuration using
* the CLUSTER SLOTS command against the specified node or a random one from
* the pool.
*
* @param NodeConnectionInterface $connection Optional connection instance.
*/
public function askSlotMap(NodeConnectionInterface $connection = null)
{
if (!$connection && !$connection = $this->getRandomConnection()) {
return;
}
$this->slotmap->reset();
$response = $this->queryClusterNodeForSlotMap($connection);
foreach ($response as $slots) {
// We only support master servers for now, so we ignore subsequent
// elements in the $slots array identifying slaves.
[$start, $end, $master] = $slots;
if ($master[0] === '') {
$this->slotmap->setSlots($start, $end, (string) $connection);
} else {
$this->slotmap->setSlots($start, $end, "{$master[0]}:{$master[1]}");
}
}
}
/**
* Guesses the correct node associated to a given slot using a precalculated
* slots map, falling back to the same logic used by Redis to initialize a
* cluster (best-effort).
*
* @param int $slot Slot index.
*
* @return string Connection ID.
*/
protected function guessNode($slot)
{
if (!$this->pool) {
throw new ClientException('No connections available in the pool');
}
if ($this->slotmap->isEmpty()) {
$this->buildSlotMap();
}
if ($node = $this->slotmap[$slot]) {
return $node;
}
$count = count($this->pool);
$index = min((int) ($slot / (int) (16384 / $count)), $count - 1);
$nodes = array_keys($this->pool);
return $nodes[$index];
}
/**
* Creates a new connection instance from the given connection ID.
*
* @param string $connectionID Identifier for the connection.
*
* @return NodeConnectionInterface
*/
protected function createConnection($connectionID)
{
$separator = strrpos($connectionID, ':');
return $this->connections->create([
'host' => substr($connectionID, 0, $separator),
'port' => substr($connectionID, $separator + 1),
]);
}
/**
* {@inheritdoc}
*/
public function getConnectionByCommand(CommandInterface $command)
{
$slot = $this->strategy->getSlot($command);
if (!isset($slot)) {
throw new NotSupportedException(
"Cannot use '{$command->getId()}' with redis-cluster."
);
}
if (isset($this->slots[$slot])) {
return $this->slots[$slot];
} else {
return $this->getConnectionBySlot($slot);
}
}
/**
* Returns the connection currently associated to a given slot.
*
* @param int $slot Slot index.
*
* @return NodeConnectionInterface
* @throws OutOfBoundsException
*/
public function getConnectionBySlot($slot)
{
if (!SlotMap::isValid($slot)) {
throw new OutOfBoundsException("Invalid slot [$slot].");
}
if (isset($this->slots[$slot])) {
return $this->slots[$slot];
}
$connectionID = $this->guessNode($slot);
if (!$connection = $this->getConnectionById($connectionID)) {
$connection = $this->createConnection($connectionID);
$this->pool[$connectionID] = $connection;
}
return $this->slots[$slot] = $connection;
}
/**
* {@inheritdoc}
*/
public function getConnectionById($connectionID)
{
return $this->pool[$connectionID] ?? null;
}
/**
* Returns a random connection from the pool.
*
* @return NodeConnectionInterface|null
*/
protected function getRandomConnection()
{
if (!$this->pool) {
return null;
}
return $this->pool[array_rand($this->pool)];
}
/**
* Permanently associates the connection instance to a new slot.
* The connection is added to the connections pool if not yet included.
*
* @param NodeConnectionInterface $connection Connection instance.
* @param int $slot Target slot index.
*/
protected function move(NodeConnectionInterface $connection, $slot)
{
$this->pool[(string) $connection] = $connection;
$this->slots[(int) $slot] = $connection;
$this->slotmap[(int) $slot] = $connection;
}
/**
* Handles -ERR responses returned by Redis.
*
* @param CommandInterface $command Command that generated the -ERR response.
* @param ErrorResponseInterface $error Redis error response object.
*
* @return mixed
*/
protected function onErrorResponse(CommandInterface $command, ErrorResponseInterface $error)
{
$details = explode(' ', $error->getMessage(), 2);
switch ($details[0]) {
case 'MOVED':
return $this->onMovedResponse($command, $details[1]);
case 'ASK':
return $this->onAskResponse($command, $details[1]);
default:
return $error;
}
}
/**
* Handles -MOVED responses by executing again the command against the node
* indicated by the Redis response.
*
* @param CommandInterface $command Command that generated the -MOVED response.
* @param string $details Parameters of the -MOVED response.
*
* @return mixed
*/
protected function onMovedResponse(CommandInterface $command, $details)
{
[$slot, $connectionID] = explode(' ', $details, 2);
if (!$connection = $this->getConnectionById($connectionID)) {
$connection = $this->createConnection($connectionID);
}
if ($this->useClusterSlots) {
$this->askSlotMap($connection);
}
$this->move($connection, $slot);
return $this->executeCommand($command);
}
/**
* Handles -ASK responses by executing again the command against the node
* indicated by the Redis response.
*
* @param CommandInterface $command Command that generated the -ASK response.
* @param string $details Parameters of the -ASK response.
*
* @return mixed
*/
protected function onAskResponse(CommandInterface $command, $details)
{
[$slot, $connectionID] = explode(' ', $details, 2);
if (!$connection = $this->getConnectionById($connectionID)) {
$connection = $this->createConnection($connectionID);
}
$connection->executeCommand(RawCommand::create('ASKING'));
return $connection->executeCommand($command);
}
/**
* Ensures that a command is executed one more time on connection failure.
*
* The connection to the node that generated the error is evicted from the
* pool before trying to fetch an updated slots map from another node. If
* the new slots map points to an unreachable server the client gives up and
* throws the exception as the nodes participating in the cluster may still
* have to agree that something changed in the configuration of the cluster.
*
* @param CommandInterface $command Command instance.
* @param string $method Actual method.
*
* @return mixed
*/
private function retryCommandOnFailure(CommandInterface $command, $method)
{
$retries = 0;
$retryAfter = $this->retryInterval;
while ($retries <= $this->retryLimit) {
try {
$response = $this->getConnectionByCommand($command)->$method($command);
if ($response instanceof ErrorResponse) {
$message = $response->getMessage();
if (strpos($message, 'CLUSTERDOWN') !== false) {
throw new ServerException($message);
}
}
break;
} catch (Throwable $exception) {
usleep($retryAfter * 1000);
$retryAfter = $retryAfter * 2;
if ($exception instanceof ConnectionException) {
$connection = $exception->getConnection();
if ($connection) {
$connection->disconnect();
$this->remove($connection);
}
}
if ($retries === $this->retryLimit) {
throw $exception;
}
if ($this->useClusterSlots) {
$this->askSlotMap();
}
++$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)
{
$response = $this->retryCommandOnFailure($command, __FUNCTION__);
if ($response instanceof ErrorResponseInterface) {
return $this->onErrorResponse($command, $response);
}
return $response;
}
/**
* @return int
*/
#[ReturnTypeWillChange]
public function count()
{
return count($this->pool);
}
/**
* @return Traversable<string, NodeConnectionInterface>
*/
#[ReturnTypeWillChange]
public function getIterator()
{
if ($this->slotmap->isEmpty()) {
$this->useClusterSlots ? $this->askSlotMap() : $this->buildSlotMap();
}
$connections = [];
foreach ($this->slotmap->getNodes() as $node) {
if (!$connection = $this->getConnectionById($node)) {
$this->add($connection = $this->createConnection($node));
}
$connections[] = $connection;
}
return new ArrayIterator($connections);
}
/**
* Returns the underlying slot map.
*
* @return SlotMap
*/
public function getSlotMap()
{
return $this->slotmap;
}
/**
* Returns the underlying command hash strategy used to hash commands by
* using keys found in their arguments.
*
* @return StrategyInterface
*/
public function getClusterStrategy()
{
return $this->strategy;
}
/**
* Returns the underlying connection factory used to create new connection
* instances to Redis nodes indicated by redis-cluster.
*
* @return FactoryInterface
*/
public function getConnectionFactory()
{
return $this->connections;
}
/**
* Enables automatic fetching of the current slots map from one of the nodes
* using the CLUSTER SLOTS command. This option is enabled by default as
* asking the current slots map to Redis upon -MOVED responses may reduce
* overhead by eliminating the trial-and-error nature of the node guessing
* procedure, mostly when targeting many keys that would end up in a lot of
* redirections.
*
* The slots map can still be manually fetched using the askSlotMap()
* method whether or not this option is enabled.
*
* @param bool $value Enable or disable the use of CLUSTER SLOTS.
*/
public function useClusterSlots($value)
{
$this->useClusterSlots = (bool) $value;
}
}

337
plugins/cache-redis/Predis/Connection/RelayConnection.php Normal file
View file

@ -0,0 +1,337 @@
<?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 InvalidArgumentException;
use Predis\ClientException;
use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Response\ServerException;
use Relay\Exception as RelayException;
use Relay\Relay;
/**
* This class provides the implementation of a Predis connection that
* uses Relay for network communication and in-memory caching.
*
* Using Relay allows for:
* 1) significantly faster reads thanks to in-memory caching
* 2) fast data serialization using igbinary
* 3) fast data compression using lzf, lz4 or zstd
*
* Usage of igbinary serialization and zstd compresses reduces
* network traffic and Redis memory usage by ~75%.
*
* For instructions on how to install the Relay extension, please consult
* the repository of the project: https://relay.so/docs/installation
*
* The connection parameters supported by this class are:
*
* - scheme: it can be either 'tcp', 'tls' 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.
* - read_write_timeout: timeout of read / write operations.
* - cache: whether to use in-memory caching
* - serializer: data serializer
* - compression: data compression algorithm
*
* @see https://github.com/cachewerk/relay
*/
class RelayConnection extends StreamConnection
{
use RelayMethods;
/**
* The Relay instance.
*
* @var \Relay\Relay
*/
protected $client;
/**
* These commands must be called on the client, not using `Relay::rawCommand()`.
*
* @var string[]
*/
public $atypicalCommands = [
'AUTH',
'SELECT',
'TYPE',
'MULTI',
'EXEC',
'DISCARD',
'WATCH',
'UNWATCH',
'SUBSCRIBE',
'UNSUBSCRIBE',
'PSUBSCRIBE',
'PUNSUBSCRIBE',
'SSUBSCRIBE',
'SUNSUBSCRIBE',
];
/**
* {@inheritdoc}
*/
public function __construct(ParametersInterface $parameters)
{
$this->assertExtensions();
$this->parameters = $this->assertParameters($parameters);
$this->client = $this->createClient();
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
return $this->client->isConnected();
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
if ($this->client->isConnected()) {
$this->client->close();
}
}
/**
* Checks if the Relay extension is loaded in PHP.
*/
private function assertExtensions()
{
if (!extension_loaded('relay')) {
throw new NotSupportedException(
'The "relay" extension is required by this connection backend.'
);
}
}
/**
* {@inheritdoc}
*/
protected function assertParameters(ParametersInterface $parameters)
{
if (!in_array($parameters->scheme, ['tcp', 'tls', 'unix', 'redis', 'rediss'])) {
throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
}
if (!in_array($parameters->serializer, [null, 'php', 'igbinary', 'msgpack', 'json'])) {
throw new InvalidArgumentException("Invalid serializer: '{$parameters->serializer}'.");
}
if (!in_array($parameters->compression, [null, 'lzf', 'lz4', 'zstd'])) {
throw new InvalidArgumentException("Invalid compression algorithm: '{$parameters->compression}'.");
}
return $parameters;
}
/**
* Creates a new instance of the client.
*
* @return \Relay\Relay
*/
private function createClient()
{
$client = new Relay();
// throw when errors occur and return `null` for non-existent keys
$client->setOption(Relay::OPT_PHPREDIS_COMPATIBILITY, false);
// use reply literals
$client->setOption(Relay::OPT_REPLY_LITERAL, true);
// disable Relay's command/connection retry
$client->setOption(Relay::OPT_MAX_RETRIES, 0);
// whether to use in-memory caching
$client->setOption(Relay::OPT_USE_CACHE, $this->parameters->cache ?? true);
// set data serializer
$client->setOption(Relay::OPT_SERIALIZER, constant(sprintf(
'%s::SERIALIZER_%s',
Relay::class,
strtoupper($this->parameters->serializer ?? 'none')
)));
// set data compression algorithm
$client->setOption(Relay::OPT_COMPRESSION, constant(sprintf(
'%s::COMPRESSION_%s',
Relay::class,
strtoupper($this->parameters->compression ?? 'none')
)));
return $client;
}
/**
* Returns the underlying client.
*
* @return \Relay\Relay
*/
public function getClient()
{
return $this->client;
}
/**
* {@inheritdoc}
*/
protected function getIdentifier()
{
return $this->client->endpointId();
}
/**
* {@inheritdoc}
*/
protected function createStreamSocket(ParametersInterface $parameters, $address, $flags)
{
$timeout = isset($parameters->timeout) ? (float) $parameters->timeout : 5.0;
$retry_interval = 0;
$read_timeout = 5.0;
if (isset($parameters->read_write_timeout)) {
$read_timeout = (float) $parameters->read_write_timeout;
$read_timeout = $read_timeout > 0 ? $read_timeout : 0;
}
try {
$this->client->connect(
$parameters->path ?? $parameters->host,
isset($parameters->path) ? 0 : $parameters->port,
$timeout,
null,
$retry_interval,
$read_timeout
);
} catch (RelayException $ex) {
$this->onConnectionError($ex->getMessage(), $ex->getCode());
}
return $this->client;
}
/**
* {@inheritdoc}
*/
public function executeCommand(CommandInterface $command)
{
if (!$this->client->isConnected()) {
$this->getResource();
}
try {
$name = $command->getId();
// When using compression or a serializer, we'll need a dedicated
// handler for `Predis\Command\RawCommand` calls, currently both
// parameters are unsupported until a future Relay release
return in_array($name, $this->atypicalCommands)
? $this->client->{$name}(...$command->getArguments())
: $this->client->rawCommand($name, ...$command->getArguments());
} catch (RelayException $ex) {
throw $this->onCommandError($ex, $command);
}
}
/**
* {@inheritdoc}
*/
public function onCommandError(RelayException $exception, CommandInterface $command)
{
$code = $exception->getCode();
$message = $exception->getMessage();
if (strpos($message, 'RELAY_ERR_IO')) {
return new ConnectionException($this, $message, $code, $exception);
}
if (strpos($message, 'RELAY_ERR_REDIS')) {
return new ServerException($message, $code, $exception);
}
if (strpos($message, 'RELAY_ERR_WRONGTYPE') && strpos($message, "Got reply-type 'status'")) {
$message = 'Operation against a key holding the wrong kind of value';
}
return new ClientException($message, $code, $exception);
}
/**
* Applies the configured serializer and compression to given value.
*
* @param mixed $value
* @return string
*/
public function pack($value)
{
return $this->client->_pack($value);
}
/**
* Deserializes and decompresses to given value.
*
* @param mixed $value
* @return string
*/
public function unpack($value)
{
return $this->client->_unpack($value);
}
/**
* {@inheritdoc}
*/
public function writeRequest(CommandInterface $command)
{
throw new NotSupportedException('The "relay" extension does not support writing requests.');
}
/**
* {@inheritdoc}
*/
public function readResponse(CommandInterface $command)
{
throw new NotSupportedException('The "relay" extension does not support reading responses.');
}
/**
* {@inheritdoc}
*/
public function __destruct()
{
$this->disconnect();
}
/**
* {@inheritdoc}
*/
public function __wakeup()
{
$this->assertExtensions();
$this->client = $this->createClient();
}
}

136
plugins/cache-redis/Predis/Connection/RelayMethods.php Normal file
View file

@ -0,0 +1,136 @@
<?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;
trait RelayMethods
{
/**
* Registers a new `flushed` event listener.
*
* @param callable $callback
* @return bool
*/
public function onFlushed(?callable $callback)
{
return $this->client->onFlushed($callback);
}
/**
* Registers a new `invalidated` event listener.
*
* @param callable $callback
* @param string $pattern
* @return bool
*/
public function onInvalidated(?callable $callback, string $pattern = null)
{
return $this->client->onInvalidated($callback, $pattern);
}
/**
* Dispatches all pending events.
*
* @return int|false
*/
public function dispatchEvents()
{
return $this->client->dispatchEvents();
}
/**
* Adds ignore pattern(s). Matching keys will not be cached in memory.
*
* @param string $pattern,...
* @return int
*/
public function addIgnorePatterns(string ...$pattern)
{
return $this->client->addIgnorePatterns(...$pattern);
}
/**
* Adds allow pattern(s). Only matching keys will be cached in memory.
*
* @param string $pattern,...
* @return int
*/
public function addAllowPatterns(string ...$pattern)
{
return $this->client->addAllowPatterns(...$pattern);
}
/**
* Returns the connection's endpoint identifier.
*
* @return string|false
*/
public function endpointId()
{
return $this->client->endpointId();
}
/**
* Returns a unique representation of the underlying socket connection identifier.
*
* @return string|false
*/
public function socketId()
{
return $this->client->socketId();
}
/**
* Returns information about the license.
*
* @return array<string, mixed>
*/
public function license()
{
return $this->client->license();
}
/**
* Returns statistics about Relay.
*
* @return array<string, array<string, mixed>>
*/
public function stats()
{
return $this->client->stats();
}
/**
* Returns the number of bytes allocated, or `0` in client-only mode.
*
* @return int
*/
public function maxMemory()
{
return $this->client->maxMemory();
}
/**
* Flushes Relay's in-memory cache of all databases.
* When given an endpoint, only that connection will be flushed.
* When given an endpoint and database index, only that database
* for that connection will be flushed.
*
* @param ?string $endpointId
* @param ?int $db
* @return bool
*/
public function flushMemory(string $endpointId = null, int $db = null)
{
return $this->client->flushMemory($endpointId, $db);
}
}

553
plugins/cache-redis/Predis/Connection/Replication/MasterSlaveReplication.php Normal file
View file

@ -0,0 +1,553 @@
<?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\Replication;
use InvalidArgumentException;
use Predis\ClientException;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Replication\MissingMasterException;
use Predis\Replication\ReplicationStrategy;
use Predis\Response\ErrorInterface as ResponseErrorInterface;
/**
* Aggregate connection handling replication of Redis nodes configured in a
* single master / multiple slaves setup.
*/
class MasterSlaveReplication implements ReplicationInterface
{
/**
* @var ReplicationStrategy
*/
protected $strategy;
/**
* @var NodeConnectionInterface
*/
protected $master;
/**
* @var NodeConnectionInterface[]
*/
protected $slaves = [];
/**
* @var NodeConnectionInterface[]
*/
protected $pool = [];
/**
* @var NodeConnectionInterface[]
*/
protected $aliases = [];
/**
* @var NodeConnectionInterface
*/
protected $current;
/**
* @var bool
*/
protected $autoDiscovery = false;
/**
* @var FactoryInterface
*/
protected $connectionFactory;
/**
* {@inheritdoc}
*/
public function __construct(ReplicationStrategy $strategy = null)
{
$this->strategy = $strategy ?: new ReplicationStrategy();
}
/**
* Configures the automatic discovery of the replication configuration on failure.
*
* @param bool $value Enable or disable auto discovery.
*/
public function setAutoDiscovery($value)
{
if (!$this->connectionFactory) {
throw new ClientException('Automatic discovery requires a connection factory');
}
$this->autoDiscovery = (bool) $value;
}
/**
* Sets the connection factory used to create the connections by the auto
* discovery procedure.
*
* @param FactoryInterface $connectionFactory Connection factory instance.
*/
public function setConnectionFactory(FactoryInterface $connectionFactory)
{
$this->connectionFactory = $connectionFactory;
}
/**
* Resets the connection state.
*/
protected function reset()
{
$this->current = null;
}
/**
* {@inheritdoc}
*/
public function add(NodeConnectionInterface $connection)
{
$parameters = $connection->getParameters();
if ('master' === $parameters->role) {
$this->master = $connection;
} else {
// everything else is considered a slvave.
$this->slaves[] = $connection;
}
if (isset($parameters->alias)) {
$this->aliases[$parameters->alias] = $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]);
} else {
return false;
}
unset($this->pool[(string) $connection]);
if ($this->aliases && $alias = $connection->getParameters()->alias) {
unset($this->aliases[$alias]);
}
$this->reset();
return true;
}
/**
* {@inheritdoc}
*/
public function getConnectionByCommand(CommandInterface $command)
{
if (!$this->current) {
if ($this->strategy->isReadOperation($command) && $slave = $this->pickSlave()) {
$this->current = $slave;
} else {
$this->current = $this->getMasterOrDie();
}
return $this->current;
}
if ($this->current === $master = $this->getMasterOrDie()) {
return $master;
}
if (!$this->strategy->isReadOperation($command) || !$this->slaves) {
$this->current = $master;
}
return $this->current;
}
/**
* {@inheritdoc}
*/
public function getConnectionById($id)
{
return $this->pool[$id] ?? null;
}
/**
* Returns a connection instance by its alias.
*
* @param string $alias Connection alias.
*
* @return NodeConnectionInterface|null
*/
public function getConnectionByAlias($alias)
{
return $this->aliases[$alias] ?? null;
}
/**
* Returns a connection by its role.
*
* @param string $role Connection role (`master` or `slave`)
*
* @return NodeConnectionInterface|null
*/
public function getConnectionByRole($role)
{
if ($role === 'master') {
return $this->getMaster();
} elseif ($role === 'slave') {
return $this->pickSlave();
}
return null;
}
/**
* Switches the internal connection in use by the backend.
*
* @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.');
}
$this->current = $connection;
}
/**
* {@inheritdoc}
*/
public function switchToMaster()
{
if (!$connection = $this->getConnectionByRole('master')) {
throw new InvalidArgumentException('Invalid connection or connection not found.');
}
$this->switchTo($connection);
}
/**
* {@inheritdoc}
*/
public function switchToSlave()
{
if (!$connection = $this->getConnectionByRole('slave')) {
throw new InvalidArgumentException('Invalid connection or connection not found.');
}
$this->switchTo($connection);
}
/**
* {@inheritdoc}
*/
public function getCurrent()
{
return $this->current;
}
/**
* {@inheritdoc}
*/
public function getMaster()
{
return $this->master;
}
/**
* Returns the connection associated to the master server.
*
* @return NodeConnectionInterface
*/
private function getMasterOrDie()
{
if (!$connection = $this->getMaster()) {
throw new MissingMasterException('No master server available for replication');
}
return $connection;
}
/**
* {@inheritdoc}
*/
public function getSlaves()
{
return $this->slaves;
}
/**
* Returns the underlying replication strategy.
*
* @return ReplicationStrategy
*/
public function getReplicationStrategy()
{
return $this->strategy;
}
/**
* Returns a random slave.
*
* @return NodeConnectionInterface|null
*/
protected function pickSlave()
{
if (!$this->slaves) {
return null;
}
return $this->slaves[array_rand($this->slaves)];
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
return $this->current ? $this->current->isConnected() : false;
}
/**
* {@inheritdoc}
*/
public function connect()
{
if (!$this->current) {
if (!$this->current = $this->pickSlave()) {
if (!$this->current = $this->getMaster()) {
throw new ClientException('No available connection for replication');
}
}
}
$this->current->connect();
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
foreach ($this->pool as $connection) {
$connection->disconnect();
}
}
/**
* Handles response from INFO.
*
* @param string $response
*
* @return array
*/
private function handleInfoResponse($response)
{
$info = [];
foreach (preg_split('/\r?\n/', $response) as $row) {
if (strpos($row, ':') === false) {
continue;
}
[$k, $v] = explode(':', $row, 2);
$info[$k] = $v;
}
return $info;
}
/**
* Fetches the replication configuration from one of the servers.
*/
public function discover()
{
if (!$this->connectionFactory) {
throw new ClientException('Discovery requires a connection factory');
}
while (true) {
try {
if ($connection = $this->getMaster()) {
$this->discoverFromMaster($connection, $this->connectionFactory);
break;
} elseif ($connection = $this->pickSlave()) {
$this->discoverFromSlave($connection, $this->connectionFactory);
break;
} else {
throw new ClientException('No connection available for discovery');
}
} catch (ConnectionException $exception) {
$this->remove($connection);
}
}
}
/**
* Discovers the replication configuration by contacting the master node.
*
* @param NodeConnectionInterface $connection Connection to the master node.
* @param FactoryInterface $connectionFactory Connection factory instance.
*/
protected function discoverFromMaster(NodeConnectionInterface $connection, FactoryInterface $connectionFactory)
{
$response = $connection->executeCommand(RawCommand::create('INFO', 'REPLICATION'));
$replication = $this->handleInfoResponse($response);
if ($replication['role'] !== 'master') {
throw new ClientException("Role mismatch (expected master, got slave) [$connection]");
}
$this->slaves = [];
foreach ($replication as $k => $v) {
$parameters = null;
if (strpos($k, 'slave') === 0 && preg_match('/ip=(?P<host>.*),port=(?P<port>\d+)/', $v, $parameters)) {
$slaveConnection = $connectionFactory->create([
'host' => $parameters['host'],
'port' => $parameters['port'],
'role' => 'slave',
]);
$this->add($slaveConnection);
}
}
}
/**
* Discovers the replication configuration by contacting one of the slaves.
*
* @param NodeConnectionInterface $connection Connection to one of the slaves.
* @param FactoryInterface $connectionFactory Connection factory instance.
*/
protected function discoverFromSlave(NodeConnectionInterface $connection, FactoryInterface $connectionFactory)
{
$response = $connection->executeCommand(RawCommand::create('INFO', 'REPLICATION'));
$replication = $this->handleInfoResponse($response);
if ($replication['role'] !== 'slave') {
throw new ClientException("Role mismatch (expected slave, got master) [$connection]");
}
$masterConnection = $connectionFactory->create([
'host' => $replication['master_host'],
'port' => $replication['master_port'],
'role' => 'master',
]);
$this->add($masterConnection);
$this->discoverFromMaster($masterConnection, $connectionFactory);
}
/**
* Retries the execution of a command upon slave failure.
*
* @param CommandInterface $command Command instance.
* @param string $method Actual method.
*
* @return mixed
*/
private function retryCommandOnFailure(CommandInterface $command, $method)
{
while (true) {
try {
$connection = $this->getConnectionByCommand($command);
$response = $connection->$method($command);
if ($response instanceof ResponseErrorInterface && $response->getErrorType() === 'LOADING') {
throw new ConnectionException($connection, "Redis is loading the dataset in memory [$connection]");
}
break;
} catch (ConnectionException $exception) {
$connection = $exception->getConnection();
$connection->disconnect();
if ($connection === $this->master && !$this->autoDiscovery) {
// Throw immediately when master connection is failing, even
// when the command represents a read-only operation, unless
// automatic discovery has been enabled.
throw $exception;
} else {
// Otherwise remove the failing slave and attempt to execute
// the command again on one of the remaining slaves...
$this->remove($connection);
}
// ... that is, unless we have no more connections to use.
if (!$this->slaves && !$this->master) {
throw $exception;
} elseif ($this->autoDiscovery) {
$this->discover();
}
} catch (MissingMasterException $exception) {
if ($this->autoDiscovery) {
$this->discover();
} else {
throw $exception;
}
}
}
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__);
}
/**
* {@inheritdoc}
*/
public function __sleep()
{
return ['master', 'slaves', 'pool', 'aliases', 'strategy'];
}
}

53
plugins/cache-redis/Predis/Connection/Replication/ReplicationInterface.php Normal file
View file

@ -0,0 +1,53 @@
<?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\Replication;
use Predis\Connection\AggregateConnectionInterface;
use Predis\Connection\NodeConnectionInterface;
/**
* Defines a group of Redis nodes in a master / slave replication setup.
*/
interface ReplicationInterface extends AggregateConnectionInterface
{
/**
* Switches the internal connection in use to the master server.
*/
public function switchToMaster();
/**
* Switches the internal connection in use to a random slave server.
*/
public function switchToSlave();
/**
* Returns the connection in use by the replication backend.
*
* @return NodeConnectionInterface
*/
public function getCurrent();
/**
* Returns the connection to the master server.
*
* @return NodeConnectionInterface
*/
public function getMaster();
/**
* Returns a list of connections to slave servers.
*
* @return NodeConnectionInterface[]
*/
public function getSlaves();
}

775
plugins/cache-redis/Predis/Connection/Replication/SentinelReplication.php Normal file
View file

@ -0,0 +1,775 @@
<?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\Replication;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\CommunicationException;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface as ConnectionFactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Connection\Parameters;
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 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 $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 AUTH and SELECT commands so we must
// explicitly set them to NULL to avoid problems when using default
// parameters set via client options. Actually AUTH is supported for
// sentinels starting with Redis 5 but we have to differentiate from
// sentinels passwords and nodes passwords, this will be implemented
// in a later release.
$parameters['database'] = null;
$parameters['username'] = 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;
}
$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',
];
}
}