123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199 |
- <?php
- namespace Predis\Pipeline;
- use Predis\Client;
- use Predis\Helpers;
- use Predis\ClientException;
- use Predis\Commands\ICommand;
- class PipelineContext
- {
- private $client;
- private $executor;
- private $pipeline = array();
- private $replies = array();
- private $running = false;
-
- public function __construct(Client $client, Array $options = null)
- {
- $this->client = $client;
- $this->executor = $this->createExecutor($client, $options ?: array());
- }
-
- protected function createExecutor(Client $client, Array $options)
- {
- if (!$options) {
- return new StandardExecutor();
- }
- if (isset($options['executor'])) {
- $executor = $options['executor'];
- if (!$executor instanceof IPipelineExecutor) {
- throw new \InvalidArgumentException(
- 'The executor option accepts only instances ' .
- 'of Predis\Pipeline\IPipelineExecutor'
- );
- }
- return $executor;
- }
- if (isset($options['safe']) && $options['safe'] == true) {
- $isCluster = Helpers::isCluster($client->getConnection());
- return $isCluster ? new SafeClusterExecutor() : new SafeExecutor();
- }
- return new StandardExecutor();
- }
-
- public function __call($method, $arguments)
- {
- $command = $this->client->createCommand($method, $arguments);
- $this->recordCommand($command);
- return $this;
- }
-
- protected function recordCommand(ICommand $command)
- {
- $this->pipeline[] = $command;
- }
-
- public function executeCommand(ICommand $command)
- {
- $this->recordCommand($command);
- }
-
- public function flushPipeline($send = true)
- {
- if (count($this->pipeline) > 0) {
- if ($send) {
- $connection = $this->client->getConnection();
- $replies = $this->executor->execute($connection, $this->pipeline);
- $this->replies = array_merge($this->replies, $replies);
- }
- $this->pipeline = array();
- }
- return $this;
- }
-
- private function setRunning($bool)
- {
- if ($bool === true && $this->running === true) {
- throw new ClientException("This pipeline is already opened");
- }
- $this->running = $bool;
- }
-
- public function execute($callable = null)
- {
- if ($callable && !is_callable($callable)) {
- throw new \InvalidArgumentException('Argument passed must be a callable object');
- }
- $this->setRunning(true);
- $pipelineBlockException = null;
- try {
- if ($callable !== null) {
- call_user_func($callable, $this);
- }
- $this->flushPipeline();
- }
- catch (\Exception $exception) {
- $pipelineBlockException = $exception;
- }
- $this->setRunning(false);
- if ($pipelineBlockException !== null) {
- throw $pipelineBlockException;
- }
- return $this->replies;
- }
-
- public function getClient()
- {
- return $this->client;
- }
-
- public function getExecutor()
- {
- return $this->executor;
- }
- }
|