forked from bigcommerce/predis
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge remote-tracking branch 'github/pr/656' into main
Local branch v2.0-clientoptions-aggregateconnections
- Loading branch information
Showing
11 changed files
with
892 additions
and
251 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,56 +11,99 @@ | |
|
|
||
| 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 <[email protected]> | ||
| */ | ||
| 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; | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * {@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)); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.