* @return mixed * @throws ServerException */ protected function onErrorResponse(CommandInterface $command, ErrorResponseInterface $response) { if ($command instanceof ScriptCommand && $response->getErrorType() === 'NOSCRIPT') { $response = $this->executeCommand($command->getEvalCommand()); if (!$response instanceof ResponseInterface) { $response = $command->parseResponse($response); } return $response; } if ($this->options->exceptions) { throw new ServerException($response->getMessage()); } return $response; } /** * Executes the specified initializer method on `$this` by adjusting the * actual invocation depending on the arity (0, 1 or 2 arguments). This is * simply an utility method to create Redis contexts instances since they * follow a common initialization path. * * @param string $initializer Method name. * @param array $argv Arguments for the method. * * @return mixed */ private function sharedContextFactory($initializer, $argv = null) { switch (count($argv)) { case 0: return $this->$initializer(); case 1: return is_array($argv[0]) ? $this->$initializer($argv[0]) : $this->$initializer(null, $argv[0]); case 2: [$arg0, $arg1] = $argv; return $this->$initializer($arg0, $arg1); default: return $this->$initializer($this, $argv); } } /** * Creates a new pipeline context and returns it, or returns the results of * a pipeline executed inside the optionally provided callable object. * * @param mixed ...$arguments Array of options, a callable for execution, or both. * * @return Pipeline|array */ public function pipeline(...$arguments) { return $this->sharedContextFactory('createPipeline', func_get_args()); } /** * Actual pipeline context initializer method. * * @param array|null $options Options for the context. * @param mixed $callable Optional callable used to execute the context. * * @return Pipeline|array */ protected function createPipeline(?array $options = null, $callable = null) { if (isset($options['atomic']) && $options['atomic']) { $class = Atomic::class; } elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) { $class = FireAndForget::class; } else { $class = Pipeline::class; } if ($this->connection instanceof RelayConnection) { if (isset($options['atomic']) && $options['atomic']) { $class = RelayAtomic::class; } elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) { throw new NotSupportedException('The "relay" extension does not support fire-and-forget pipelines.'); } else { $class = RelayPipeline::class; } } /* * @var ClientContextInterface */ $pipeline = new $class($this); if (isset($callable)) { return $pipeline->execute($callable); } return $pipeline; } /** * Creates a new transaction context and returns it, or returns the results * of a transaction executed inside the optionally provided callable object. * * @param mixed ...$arguments Array of options, a callable for execution, or both. * * @return MultiExecTransaction|array */ public function transaction(...$arguments) { return $this->sharedContextFactory('createTransaction', func_get_args()); } /** * Actual transaction context initializer method. * * @param array|null $options Options for the context. * @param mixed $callable Optional callable used to execute the context. * * @return MultiExecTransaction|array */ protected function createTransaction(?array $options = null, $callable = null) { $transaction = new MultiExecTransaction($this, $options); if (isset($callable)) { return $transaction->execute($callable); } return $transaction; } /** * Creates a new publish/subscribe context and returns it, or starts its loop * inside the optionally provided callable object. * * @param mixed ...$arguments Array of options, a callable for execution, or both. * * @return PubSubConsumer|null */ public function pubSubLoop(...$arguments) { return $this->sharedContextFactory('createPubSub', func_get_args()); } /** * Actual publish/subscribe context initializer method. * * @param array|null $options Options for the context. * @param mixed $callable Optional callable used to execute the context. * * @return PubSubConsumer|null */ protected function createPubSub(?array $options = null, $callable = null) { if ($this->connection instanceof RelayConnection) { $pubsub = new RelayPubSubConsumer($this, $options); } else { $pubsub = new PubSubConsumer($this, $options); } if (!isset($callable)) { return $pubsub; } foreach ($pubsub as $message) { if (call_user_func($callable, $pubsub, $message) === false) { $pubsub->stop(); } } return null; } /** * Creates a new monitor consumer and returns it. * * @return MonitorConsumer */ public function monitor() { return new MonitorConsumer($this); } /** * @return Traversable */ #[ReturnTypeWillChange] public function getIterator() { $clients = []; $connection = $this->getConnection(); if (!$connection instanceof Traversable) { return new ArrayIterator([ (string) $connection => new static($connection, $this->getOptions()), ]); } foreach ($connection as $node) { $clients[(string) $node] = new static($node, $this->getOptions()); } return new ArrayIterator($clients); } }