123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427 |
- <?php
- namespace Predis;
- use Predis\Command\CommandInterface;
- use Predis\Command\ScriptedCommand;
- use Predis\Connection\AggregatedConnectionInterface;
- use Predis\Connection\ConnectionInterface;
- use Predis\Connection\ConnectionFactory;
- use Predis\Connection\ConnectionFactoryInterface;
- use Predis\Monitor\MonitorContext;
- use Predis\Option\ClientOptions;
- use Predis\Option\ClientOptionsInterface;
- use Predis\Pipeline\PipelineContext;
- use Predis\Profile\ServerProfile;
- use Predis\Profile\ServerProfileInterface;
- use Predis\PubSub\PubSubContext;
- use Predis\Transaction\MultiExecContext;
- class Client implements ClientInterface
- {
- const VERSION = '0.8.1-dev';
- private $options;
- private $profile;
- private $connection;
- private $connections;
-
- public function __construct($parameters = null, $options = null)
- {
- $this->options = $this->filterOptions($options);
- $this->profile = $this->options->profile;
- $this->connections = $this->options->connections;
- $this->connection = $this->initializeConnection($parameters);
- }
-
- protected function filterOptions($options)
- {
- if ($options === null) {
- return new ClientOptions();
- }
- if (is_array($options)) {
- return new ClientOptions($options);
- }
- if ($options instanceof ClientOptionsInterface) {
- return $options;
- }
- throw new \InvalidArgumentException("Invalid type for client options");
- }
-
- protected function initializeConnection($parameters)
- {
- if ($parameters instanceof ConnectionInterface) {
- return $parameters;
- }
- if (is_array($parameters) && isset($parameters[0])) {
- $replication = isset($this->options->replication) && $this->options->replication;
- $connection = $this->options->{$replication ? 'replication' : 'cluster'};
- return $this->connections->createAggregated($connection, $parameters);
- }
- return $this->connections->create($parameters);
- }
-
- public function getProfile()
- {
- return $this->profile;
- }
-
- public function getOptions()
- {
- return $this->options;
- }
-
- public function getConnectionFactory()
- {
- return $this->connections;
- }
-
- public function getClientFor($connectionID)
- {
- if (($connection = $this->getConnectionById($connectionID)) === null) {
- throw new \InvalidArgumentException("Invalid connection ID: '$connectionID'");
- }
- return new Client($connection, $this->options);
- }
-
- public function connect()
- {
- $this->connection->connect();
- }
-
- public function disconnect()
- {
- $this->connection->disconnect();
- }
-
- public function quit()
- {
- $this->disconnect();
- }
-
- public function isConnected()
- {
- return $this->connection->isConnected();
- }
-
- public function getConnection()
- {
- return $this->connection;
- }
-
- public function getConnectionById($connectionId)
- {
- if (!$this->connection instanceof AggregatedConnectionInterface) {
- throw new NotSupportedException('Retrieving connections by ID is supported only when using aggregated connections');
- }
- return $this->connection->getConnectionById($connectionId);
- }
-
- public function __call($method, $arguments)
- {
- $command = $this->profile->createCommand($method, $arguments);
- $response = $this->connection->executeCommand($command);
- if ($response instanceof ResponseObjectInterface) {
- if ($response instanceof ResponseErrorInterface) {
- $response = $this->onResponseError($command, $response);
- }
- return $response;
- }
- return $command->parseResponse($response);
- }
-
- public function createCommand($method, $arguments = array())
- {
- return $this->profile->createCommand($method, $arguments);
- }
-
- public function executeCommand(CommandInterface $command)
- {
- $response = $this->connection->executeCommand($command);
- if ($response instanceof ResponseObjectInterface) {
- if ($response instanceof ResponseErrorInterface) {
- $response = $this->onResponseError($command, $response);
- }
- return $response;
- }
- return $command->parseResponse($response);
- }
-
- protected function onResponseError(CommandInterface $command, ResponseErrorInterface $response)
- {
- if ($command instanceof ScriptedCommand && $response->getErrorType() === 'NOSCRIPT') {
- $eval = $this->createCommand('eval');
- $eval->setRawArguments($command->getEvalArguments());
- $response = $this->executeCommand($eval);
- if (false === $response instanceof ResponseObjectInterface) {
- $response = $command->parseResponse($response);
- }
- return $response;
- }
- if ($this->options->exceptions === true) {
- throw new ServerException($response->getMessage());
- }
- return $response;
- }
-
- private function sharedInitializer($argv, $initializer)
- {
- switch (count($argv)) {
- case 0:
- return $this->$initializer();
- case 1:
- list($arg0) = $argv;
- return is_array($arg0) ? $this->$initializer($arg0) : $this->$initializer(null, $arg0);
- case 2:
- list($arg0, $arg1) = $argv;
- return $this->$initializer($arg0, $arg1);
- default:
- return $this->$initializer($this, $argv);
- }
- }
-
- public function pipeline()
- {
- return $this->sharedInitializer(func_get_args(), 'initPipeline');
- }
-
- protected function initPipeline(Array $options = null, $callable = null)
- {
- $executor = isset($options['executor']) ? $options['executor'] : null;
- if (is_callable($executor)) {
- $executor = call_user_func($executor, $this, $options);
- }
- $pipeline = new PipelineContext($this, $executor);
- $replies = $this->pipelineExecute($pipeline, $callable);
- return $replies;
- }
-
- private function pipelineExecute(PipelineContext $pipeline, $callable)
- {
- return isset($callable) ? $pipeline->execute($callable) : $pipeline;
- }
-
- public function multiExec()
- {
- return $this->sharedInitializer(func_get_args(), 'initMultiExec');
- }
-
- protected function initMultiExec(Array $options = null, $callable = null)
- {
- $transaction = new MultiExecContext($this, $options ?: array());
- return isset($callable) ? $transaction->execute($callable) : $transaction;
- }
-
- public function pubSub()
- {
- return $this->sharedInitializer(func_get_args(), 'initPubSub');
- }
-
- protected function initPubSub(Array $options = null, $callable = null)
- {
- $pubsub = new PubSubContext($this, $options);
- if (!isset($callable)) {
- return $pubsub;
- }
- foreach ($pubsub as $message) {
- if (call_user_func($callable, $pubsub, $message) === false) {
- $pubsub->closeContext();
- }
- }
- }
-
- public function monitor()
- {
- return new MonitorContext($this);
- }
- }
|