mirror of
https://github.com/predis/predis.git
synced 2026-08-21 01:11:49 +00:00
Dramatically improve aggregate connections initialization.
This is a complete overhaul of how aggregate connections are created and initialized, now everything is self-contained in our usual 3 supported client options: "aggregate", "cluster" and "replication". The usage of callables acting as connection initializerss is now more consistent through the various options. When the callable is invoked it receives 3 arguments (the original set of connection parameters passed by reference, the options container, the current option) and must return an instance of Predis\Connection\AggregateConnectionInterface otherwise an InvalidArgumentException is thrown. When using "cluster" and "replication" the returned aggregate connection is automatically populated by adding the list of nodes in $parameters, on the other hand "aggregate" skips this automatism so it is up to the user. In any case the user-supplied callable receives $parameters as a reference, setting $parameters to NULL inside the body of the callable makes the client skip automatic aggregation regardless of the option in use. In addition to this the actual procedure of adding nodes to an aggregate connection has been moved directly into the respective options instead of being spread between the client (which instead should just pass a set of parameters and get back a fully-configured aggregate connection) and the connection factory (and the scope of a connection factory is only to create new connetion instances to single Redis servers).
This commit is contained in:
Daniele Alessandri
+6
-32
@@ -18,7 +18,6 @@ use Predis\Configuration\Options;
|
||||
use Predis\Configuration\OptionsInterface;
|
||||
use Predis\Connection\ConnectionInterface;
|
||||
use Predis\Connection\ParametersInterface;
|
||||
use Predis\Connection\Replication\SentinelReplication;
|
||||
use Predis\Monitor\Consumer as MonitorConsumer;
|
||||
use Predis\Pipeline\Pipeline;
|
||||
use Predis\PubSub\Consumer as PubSubConsumer;
|
||||
@@ -126,14 +125,12 @@ class Client implements ClientInterface, \IteratorAggregate
|
||||
if (is_array($parameters)) {
|
||||
if (!isset($parameters[0])) {
|
||||
return $options->connections->create($parameters);
|
||||
}
|
||||
|
||||
if ($options->defined('cluster')) {
|
||||
return $this->createAggregateConnection($parameters, 'cluster');
|
||||
} elseif ($options->defined('replication')) {
|
||||
return $this->createAggregateConnection($parameters, 'replication');
|
||||
} elseif ($options->defined('aggregate')) {
|
||||
return $this->createAggregateConnection($parameters, 'aggregate');
|
||||
} elseif ($options->defined('cluster') && $initializer = $options->cluster) {
|
||||
return $initializer($parameters, true);
|
||||
} elseif ($options->defined('replication') && $initializer = $options->replication) {
|
||||
return $initializer($parameters, true);
|
||||
} elseif ($options->defined('aggregate') && $initializer = $options->aggregate) {
|
||||
return $initializer($parameters, false);
|
||||
} else {
|
||||
throw new \InvalidArgumentException(
|
||||
'Array of connection parameters requires `cluster`, `replication` or `aggregate` client option'
|
||||
@@ -154,29 +151,6 @@ class Client implements ClientInterface, \IteratorAggregate
|
||||
throw new \InvalidArgumentException('Invalid type for connection parameters');
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an aggregate connection.
|
||||
*
|
||||
* @param mixed $parameters Connection parameters.
|
||||
* @param string $option Option for aggregate connections (`aggregate`, `cluster`, `replication`).
|
||||
*
|
||||
* @return \Closure
|
||||
*/
|
||||
protected function createAggregateConnection($parameters, $option)
|
||||
{
|
||||
$options = $this->getOptions();
|
||||
|
||||
$initializer = $options->$option;
|
||||
$connection = $initializer($parameters);
|
||||
|
||||
// TODO: this is dirty but we must skip the redis-sentinel backend for now.
|
||||
if ($option !== 'aggregate' && !$connection instanceof SentinelReplication) {
|
||||
$options->connections->aggregate($connection, $parameters);
|
||||
}
|
||||
|
||||
return $connection;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
|
||||
@@ -11,44 +11,79 @@
|
||||
|
||||
namespace Predis\Configuration\Option;
|
||||
|
||||
use InvalidArgumentException;
|
||||
use Predis\Configuration\OptionInterface;
|
||||
use Predis\Configuration\OptionsInterface;
|
||||
use Predis\Connection\AggregateConnectionInterface;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
|
||||
/**
|
||||
* Configures an aggregate connection used for clustering
|
||||
* multiple Redis nodes using various implementations with
|
||||
* different algorithms or strategies.
|
||||
* Client option for configuring generic aggregate connections.
|
||||
*
|
||||
* The only value accepted by this option is a callable that must return a valid
|
||||
* connection instance of Predis\Connection\AggregateConnectionInterface when
|
||||
* invoked by the client to create a new aggregate connection instance.
|
||||
*
|
||||
* Creation and configuration of the aggregate connection is up to the user.
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class Aggregate implements OptionInterface
|
||||
{
|
||||
/**
|
||||
* Wraps a callable to ensure that the returned value is a valid connection.
|
||||
*
|
||||
* @param OptionsInterface $options Client options.
|
||||
* @param mixed $callable Callable initializer.
|
||||
*
|
||||
* @return \Closure
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function getConnectionInitializer(OptionsInterface $options, $callable)
|
||||
public function filter(OptionsInterface $options, $value)
|
||||
{
|
||||
if (!is_callable($callable)) {
|
||||
$class = get_called_class();
|
||||
|
||||
throw new \InvalidArgumentException("$class expects a valid callable");
|
||||
if (!is_callable($value)) {
|
||||
throw new InvalidArgumentException(sprintf(
|
||||
'%s expects a callable object acting as an aggregate connection initializer',
|
||||
static::class
|
||||
));
|
||||
}
|
||||
|
||||
$option = $this;
|
||||
return $this->getConnectionInitializer($options, $value);
|
||||
}
|
||||
|
||||
return function ($parameters = null) use ($callable, $options, $option) {
|
||||
$connection = call_user_func($callable, $options, $parameters);
|
||||
/**
|
||||
* Wraps a user-supplied callable used to create a new aggregate connection.
|
||||
*
|
||||
* When the original callable acting as a connection initializer is executed
|
||||
* by the client to create a new aggregate connection, it will receive the
|
||||
* following arguments:
|
||||
*
|
||||
* - $parameters (same as passed to Predis\Client::__construct())
|
||||
* - $options (options container, Predis\Configuration\OptionsInterface)
|
||||
* - $option (current option, Predis\Configuration\OptionInterface)
|
||||
*
|
||||
* The original callable must return a valid aggregation connection instance
|
||||
* of type Predis\Connection\AggregateConnectionInterface, this is enforced
|
||||
* by the wrapper returned by this method and an exception is thrown when
|
||||
* invalid values are returned.
|
||||
*
|
||||
* @param OptionsInterface $options Client options
|
||||
* @param callable $callable Callable initializer
|
||||
*
|
||||
* @throws InvalidArgumentException
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
protected function getConnectionInitializer(OptionsInterface $options, callable $callable)
|
||||
{
|
||||
return function ($parameters = null, $autoaggregate = false) use ($callable, $options) {
|
||||
$connection = call_user_func_array($callable, [&$parameters, $options, $this]);
|
||||
|
||||
if (!$connection instanceof AggregateConnectionInterface) {
|
||||
$class = get_class($option);
|
||||
throw new InvalidArgumentException(sprintf(
|
||||
'%s expects the supplied callable to return an instance of %s, but %s was returned',
|
||||
static::class,
|
||||
AggregateConnectionInterface::class,
|
||||
is_object($connection) ? get_class($connection) : gettype($connection)
|
||||
));
|
||||
}
|
||||
|
||||
throw new \InvalidArgumentException("$class expects a valid connection type returned by callable initializer");
|
||||
if ($parameters && $autoaggregate) {
|
||||
static::aggregate($options, $connection, $parameters);
|
||||
}
|
||||
|
||||
return $connection;
|
||||
@@ -56,11 +91,19 @@ class Aggregate implements OptionInterface
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
* Adds single connections to an aggregate connection instance.
|
||||
*
|
||||
* @param OptionsInterface $options Client options
|
||||
* @param AggregateConnectionInterface $connection Target aggregate connection
|
||||
* @param array $nodes List of nodes to be added to the target aggregate connection
|
||||
*/
|
||||
public function filter(OptionsInterface $options, $value)
|
||||
public static function aggregate(OptionsInterface $options, AggregateConnectionInterface $connection, array $nodes)
|
||||
{
|
||||
return $this->getConnectionInitializer($options, $value);
|
||||
$connections = $options->connections;
|
||||
|
||||
foreach ($nodes as $node) {
|
||||
$connection->add($node instanceof NodeConnectionInterface ? $node : $connections->create($node));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
|
||||
namespace Predis\Configuration\Option;
|
||||
|
||||
use InvalidArgumentException;
|
||||
use Predis\Cluster\RedisStrategy;
|
||||
use Predis\Configuration\OptionsInterface;
|
||||
use Predis\Connection\Cluster\PredisCluster;
|
||||
@@ -25,41 +26,65 @@ use Predis\Connection\Cluster\RedisCluster;
|
||||
*/
|
||||
class Cluster extends Aggregate
|
||||
{
|
||||
/**
|
||||
* Returns a connection initializer from a descriptive name.
|
||||
*
|
||||
* @param OptionsInterface $options Client options.
|
||||
* @param string $description Identifier of a cluster backend (`predis`, `redis`)
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
protected function getConnectionInitializerByDescription(OptionsInterface $options, $description)
|
||||
{
|
||||
if ($description === 'predis') {
|
||||
$callback = $this->getDefault($options);
|
||||
} elseif ($description === 'redis') {
|
||||
$callback = function ($options) {
|
||||
return new RedisCluster($options->connections, new RedisStrategy($options->crc16));
|
||||
};
|
||||
} else {
|
||||
throw new \InvalidArgumentException(
|
||||
'String value for the cluster option must be either `predis` or `redis`'
|
||||
);
|
||||
}
|
||||
|
||||
return $this->getConnectionInitializer($options, $callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function filter(OptionsInterface $options, $value)
|
||||
{
|
||||
if (is_string($value)) {
|
||||
return $this->getConnectionInitializerByDescription($options, $value);
|
||||
} else {
|
||||
return $this->getConnectionInitializer($options, $value);
|
||||
$value = $this->getConnectionInitializerByString($options, $value);
|
||||
}
|
||||
|
||||
if (is_callable($value)) {
|
||||
return $this->getConnectionInitializer($options, $value);
|
||||
} else {
|
||||
throw new InvalidArgumentException(sprintf(
|
||||
'%s expects either a string or a callable value, %s given',
|
||||
static::class,
|
||||
is_object($value) ? get_class($value) : gettype($value)
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a connection initializer from a descriptive name.
|
||||
*
|
||||
* @param OptionsInterface $options Client options
|
||||
* @param string $description Identifier of a replication backend (`predis`, `sentinel`)
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
protected function getConnectionInitializerByString(OptionsInterface $options, string $description)
|
||||
{
|
||||
switch ($description) {
|
||||
case 'redis':
|
||||
case 'redis-cluster':
|
||||
return function ($parameters, $options, $option) {
|
||||
return new RedisCluster($options->connections, new RedisStrategy($options->crc16));
|
||||
};
|
||||
|
||||
case 'predis':
|
||||
return $this->getDefaultConnectionInitializer();
|
||||
|
||||
default:
|
||||
throw new InvalidArgumentException(sprintf(
|
||||
'%s expects either `predis`, `redis` or `redis-cluster` as valid string values, `%s` given',
|
||||
static::class,
|
||||
$description
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the default connection initializer.
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
protected function getDefaultConnectionInitializer()
|
||||
{
|
||||
return function ($parameters, $options, $option) {
|
||||
return new PredisCluster();
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -67,8 +92,9 @@ class Cluster extends Aggregate
|
||||
*/
|
||||
public function getDefault(OptionsInterface $options)
|
||||
{
|
||||
return function ($options) {
|
||||
return new PredisCluster();
|
||||
};
|
||||
return $this->getConnectionInitializer(
|
||||
$options,
|
||||
$this->getDefaultConnectionInitializer()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,9 @@
|
||||
|
||||
namespace Predis\Configuration\Option;
|
||||
|
||||
use InvalidArgumentException;
|
||||
use Predis\Configuration\OptionsInterface;
|
||||
use Predis\Connection\AggregateConnectionInterface;
|
||||
use Predis\Connection\Replication\MasterSlaveReplication;
|
||||
use Predis\Connection\Replication\SentinelReplication;
|
||||
|
||||
@@ -23,49 +25,74 @@ use Predis\Connection\Replication\SentinelReplication;
|
||||
*/
|
||||
class Replication extends Aggregate
|
||||
{
|
||||
/**
|
||||
* Returns a connection initializer from a descriptive name.
|
||||
*
|
||||
* @param OptionsInterface $options Client options.
|
||||
* @param string $description Identifier of a replication backend (`predis`, `sentinel`)
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
protected function getConnectionInitializerByDescription(OptionsInterface $options, $description)
|
||||
{
|
||||
if ($description === 'predis') {
|
||||
$callback = $this->getDefault($options);
|
||||
} elseif ($description === 'sentinel') {
|
||||
$callback = function ($options, $sentinels) {
|
||||
return new SentinelReplication($options->service, $sentinels, $options->connections);
|
||||
};
|
||||
} else {
|
||||
throw new \InvalidArgumentException(
|
||||
'String value for the replication option must be either `predis` or `sentinel`'
|
||||
);
|
||||
}
|
||||
|
||||
return $this->getConnectionInitializer($options, $callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function filter(OptionsInterface $options, $value)
|
||||
{
|
||||
if (is_string($value)) {
|
||||
return $this->getConnectionInitializerByDescription($options, $value);
|
||||
} else {
|
||||
$value = $this->getConnectionInitializerByString($options, $value);
|
||||
}
|
||||
|
||||
if (is_callable($value)) {
|
||||
return $this->getConnectionInitializer($options, $value);
|
||||
} else {
|
||||
throw new InvalidArgumentException(sprintf(
|
||||
'%s expects either a string or a callable value, %s given',
|
||||
static::class,
|
||||
is_object($value) ? get_class($value) : gettype($value)
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
* Returns a connection initializer (callable) from a descriptive string.
|
||||
*
|
||||
* Each connection initializer is specialized for the specified replication
|
||||
* backend so that all the necessary steps for the configuration of the new
|
||||
* aggregate connection are performed inside the initializer and the client
|
||||
* receives a ready-to-use connection.
|
||||
*
|
||||
* Supported configuration values are:
|
||||
*
|
||||
* - `predis` for unmanaged replication setups
|
||||
* - `redis-sentinel` for replication setups managed by redis-sentinel
|
||||
* - `sentinel` is an alias of `redis-sentinel`
|
||||
*
|
||||
* @param OptionsInterface $options Client options
|
||||
* @param string $description Identifier of a replication backend
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
public function getDefault(OptionsInterface $options)
|
||||
protected function getConnectionInitializerByString(OptionsInterface $options, string $description)
|
||||
{
|
||||
return function ($options) {
|
||||
switch ($description) {
|
||||
case 'sentinel':
|
||||
case 'redis-sentinel':
|
||||
return function ($parameters, $options, $option) {
|
||||
return new SentinelReplication($options->service, $parameters, $options->connections);
|
||||
};
|
||||
|
||||
case 'predis':
|
||||
return $this->getDefaultConnectionInitializer($options);
|
||||
|
||||
default:
|
||||
throw new InvalidArgumentException(sprintf(
|
||||
'%s expects either `predis`, `sentinel` or `redis-sentinel` as valid string values, `%s` given',
|
||||
static::class,
|
||||
$description
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the default connection initializer.
|
||||
*
|
||||
* @return callable
|
||||
*/
|
||||
protected function getDefaultConnectionInitializer()
|
||||
{
|
||||
return function ($parameters, $options, $option) {
|
||||
$connection = new MasterSlaveReplication();
|
||||
|
||||
if ($options->autodiscovery) {
|
||||
@@ -76,4 +103,33 @@ class Replication extends Aggregate
|
||||
return $connection;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public static function aggregate(OptionsInterface $options, AggregateConnectionInterface $connection, array $nodes)
|
||||
{
|
||||
// TODO: at least for now we will replicate the previous behaviour of
|
||||
// skipping automatic aggregation when using the redis-sentinel backend
|
||||
// because $nodes contains an array of sentinel servers instead of Redis
|
||||
// servers and SentinelReplication already gets the list of sentinels in
|
||||
// the first argument of its constructor. SentinelReplication::add()
|
||||
// actually knows how to handle connections marked with role=sentinel in
|
||||
// their parameters but relying on it would require an explicit role to
|
||||
// be set by the user and I would like to avoid enforcing that for now.
|
||||
if (!$connection instanceof SentinelReplication) {
|
||||
parent::aggregate($options, $connection, $nodes);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function getDefault(OptionsInterface $options)
|
||||
{
|
||||
return $this->getConnectionInitializer(
|
||||
$options,
|
||||
$this->getDefaultConnectionInitializer()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -109,16 +109,6 @@ class Factory implements FactoryInterface
|
||||
return $connection;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function aggregate(AggregateConnectionInterface $connection, array $parameters)
|
||||
{
|
||||
foreach ($parameters as $node) {
|
||||
$connection->add($node instanceof NodeConnectionInterface ? $node : $this->create($node));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Assigns a default set of parameters applied to new connections.
|
||||
*
|
||||
|
||||
@@ -41,12 +41,4 @@ interface FactoryInterface
|
||||
* @return NodeConnectionInterface
|
||||
*/
|
||||
public function create($parameters);
|
||||
|
||||
/**
|
||||
* Aggregates single connections into an aggregate connection instance.
|
||||
*
|
||||
* @param AggregateConnectionInterface $aggregate Aggregate connection instance.
|
||||
* @param array $parameters List of parameters for each connection.
|
||||
*/
|
||||
public function aggregate(AggregateConnectionInterface $aggregate, array $parameters);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user