Client.php 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534
  1. <?php
  2. /*
  3. * This file is part of the Predis package.
  4. *
  5. * (c) Daniele Alessandri <suppakilla@gmail.com>
  6. *
  7. * For the full copyright and license information, please view the LICENSE
  8. * file that was distributed with this source code.
  9. */
  10. namespace Predis;
  11. use Predis\Command\CommandInterface;
  12. use Predis\Command\RawCommand;
  13. use Predis\Command\ScriptCommand;
  14. use Predis\Configuration\Options;
  15. use Predis\Configuration\OptionsInterface;
  16. use Predis\Connection\ConnectionInterface;
  17. use Predis\Connection\Parameters;
  18. use Predis\Connection\ParametersInterface;
  19. use Predis\Monitor\Consumer as MonitorConsumer;
  20. use Predis\Pipeline\Pipeline;
  21. use Predis\PubSub\Consumer as PubSubConsumer;
  22. use Predis\Response\ErrorInterface as ErrorResponseInterface;
  23. use Predis\Response\ResponseInterface;
  24. use Predis\Response\ServerException;
  25. use Predis\Transaction\MultiExec as MultiExecTransaction;
  26. /**
  27. * Client class used for connecting and executing commands on Redis.
  28. *
  29. * This is the main high-level abstraction of Predis upon which various other
  30. * abstractions are built. Internally it aggregates various other classes each
  31. * one with its own responsibility and scope.
  32. *
  33. * {@inheritdoc}
  34. *
  35. * @author Daniele Alessandri <suppakilla@gmail.com>
  36. */
  37. class Client implements ClientInterface, \IteratorAggregate
  38. {
  39. const VERSION = '2.0.0';
  40. /** @var OptionsInterface */
  41. private $options;
  42. /** @var ConnectionInterface */
  43. private $connection;
  44. /** @var Command\FactoryInterface */
  45. private $commands;
  46. /**
  47. * @param mixed $parameters Connection parameters for one or more servers.
  48. * @param mixed $options Options to configure some behaviours of the client.
  49. */
  50. public function __construct($parameters = null, $options = null)
  51. {
  52. $this->options = static::createOptions($options ?? new Options);
  53. $this->connection = static::createConnection($this->options, $parameters ?? new Parameters);
  54. $this->commands = $this->options->commands;
  55. }
  56. /**
  57. * Creates a new set of client options for the client.
  58. *
  59. * @param array|OptionsInterface $options Set of client options
  60. *
  61. * @throws \InvalidArgumentException
  62. *
  63. * @return OptionsInterface
  64. */
  65. protected static function createOptions($options)
  66. {
  67. if (is_array($options)) {
  68. return new Options($options);
  69. } elseif ($options instanceof OptionsInterface) {
  70. return $options;
  71. } else {
  72. throw new \InvalidArgumentException('Invalid type for client options');
  73. }
  74. }
  75. /**
  76. * Creates single or aggregate connections from supplied arguments.
  77. *
  78. * This method accepts the following types to create a connection instance:
  79. *
  80. * - Array (dictionary: single connection, indexed: aggregate connections)
  81. * - String (URI for a single connection)
  82. * - Callable (connection initializer callback)
  83. * - Instance of Predis\Connection\ParametersInterface (used as-is)
  84. * - Instance of Predis\Connection\ConnectionInterface (returned as-is)
  85. *
  86. * When a callable is passed, it receives the original set of client options
  87. * and must return an instance of Predis\Connection\ConnectionInterface.
  88. *
  89. * Connections are created using the connection factory (in case of single
  90. * connections) or a specialized aggregate connection initializer (in case
  91. * of cluster and replication) retrieved from the supplied client options.
  92. *
  93. * @param OptionsInterface $options Client options container
  94. * @param mixed $parameters Connection parameters
  95. *
  96. * @throws \InvalidArgumentException
  97. *
  98. * @return ConnectionInterface
  99. */
  100. protected static function createConnection(OptionsInterface $options, $parameters)
  101. {
  102. if ($parameters instanceof ConnectionInterface) {
  103. return $parameters;
  104. }
  105. if ($parameters instanceof ParametersInterface || is_string($parameters)) {
  106. return $options->connections->create($parameters);
  107. }
  108. if (is_array($parameters)) {
  109. if (!isset($parameters[0])) {
  110. return $options->connections->create($parameters);
  111. } elseif ($options->defined('cluster') && $initializer = $options->cluster) {
  112. return $initializer($parameters, true);
  113. } elseif ($options->defined('replication') && $initializer = $options->replication) {
  114. return $initializer($parameters, true);
  115. } elseif ($options->defined('aggregate') && $initializer = $options->aggregate) {
  116. return $initializer($parameters, false);
  117. } else {
  118. throw new \InvalidArgumentException(
  119. 'Array of connection parameters requires `cluster`, `replication` or `aggregate` client option'
  120. );
  121. }
  122. }
  123. if (is_callable($parameters)) {
  124. $connection = call_user_func($parameters, $options);
  125. if (!$connection instanceof ConnectionInterface) {
  126. throw new \InvalidArgumentException('Callable parameters must return a valid connection');
  127. }
  128. return $connection;
  129. }
  130. throw new \InvalidArgumentException('Invalid type for connection parameters');
  131. }
  132. /**
  133. * {@inheritdoc}
  134. */
  135. public function getCommandFactory()
  136. {
  137. return $this->commands;
  138. }
  139. /**
  140. * {@inheritdoc}
  141. */
  142. public function getOptions()
  143. {
  144. return $this->options;
  145. }
  146. /**
  147. * Creates a new client using a specific underlying connection.
  148. *
  149. * This method allows to create a new client instance by picking a specific
  150. * connection out of an aggregate one, with the same options of the original
  151. * client instance.
  152. *
  153. * The specified selector defines which logic to use to look for a suitable
  154. * connection by the specified value. Supported selectors are:
  155. *
  156. * - `id`
  157. * - `key`
  158. * - `slot`
  159. * - `command`
  160. * - `alias`
  161. * - `role`
  162. *
  163. * Internally the client relies on duck-typing and follows this convention:
  164. *
  165. * $selector string => getConnectionBy$selector($value) method
  166. *
  167. * This means that support for specific selectors may vary depending on the
  168. * actual logic implemented by connection classes and there is no interface
  169. * binding a connection class to implement any of these.
  170. *
  171. * @param string $selector Type of selector.
  172. * @param mixed $value Value to be used by the selector.
  173. *
  174. * @return ClientInterface
  175. */
  176. public function getClientBy($selector, $value)
  177. {
  178. $selector = strtolower($selector);
  179. if (!in_array($selector, array('id', 'key', 'slot', 'role', 'alias', 'command'))) {
  180. throw new \InvalidArgumentException("Invalid selector type: `$selector`");
  181. }
  182. if (!method_exists($this->connection, $method = "getConnectionBy$selector")) {
  183. $class = get_class($this->connection);
  184. throw new \InvalidArgumentException("Selecting connection by $selector is not supported by $class");
  185. }
  186. if (!$connection = $this->connection->$method($value)) {
  187. throw new \InvalidArgumentException("Cannot find a connection by $selector matching `$value`");
  188. }
  189. return new static($connection, $this->getOptions());
  190. }
  191. /**
  192. * Opens the underlying connection and connects to the server.
  193. */
  194. public function connect()
  195. {
  196. $this->connection->connect();
  197. }
  198. /**
  199. * Closes the underlying connection and disconnects from the server.
  200. */
  201. public function disconnect()
  202. {
  203. $this->connection->disconnect();
  204. }
  205. /**
  206. * Closes the underlying connection and disconnects from the server.
  207. *
  208. * This is the same as `Client::disconnect()` as it does not actually send
  209. * the `QUIT` command to Redis, but simply closes the connection.
  210. */
  211. public function quit()
  212. {
  213. $this->disconnect();
  214. }
  215. /**
  216. * Returns the current state of the underlying connection.
  217. *
  218. * @return bool
  219. */
  220. public function isConnected()
  221. {
  222. return $this->connection->isConnected();
  223. }
  224. /**
  225. * {@inheritdoc}
  226. */
  227. public function getConnection()
  228. {
  229. return $this->connection;
  230. }
  231. /**
  232. * Executes a command without filtering its arguments, parsing the response,
  233. * applying any prefix to keys or throwing exceptions on Redis errors even
  234. * regardless of client options.
  235. *
  236. * It is possible to identify Redis error responses from normal responses
  237. * using the second optional argument which is populated by reference.
  238. *
  239. * @param array $arguments Command arguments as defined by the command signature.
  240. * @param bool $error Set to TRUE when Redis returned an error response.
  241. *
  242. * @return mixed
  243. */
  244. public function executeRaw(array $arguments, &$error = null)
  245. {
  246. $error = false;
  247. $commandID = array_shift($arguments);
  248. $response = $this->connection->executeCommand(
  249. new RawCommand($commandID, $arguments)
  250. );
  251. if ($response instanceof ResponseInterface) {
  252. if ($response instanceof ErrorResponseInterface) {
  253. $error = true;
  254. }
  255. return (string) $response;
  256. }
  257. return $response;
  258. }
  259. /**
  260. * {@inheritdoc}
  261. */
  262. public function __call($commandID, $arguments)
  263. {
  264. return $this->executeCommand(
  265. $this->createCommand($commandID, $arguments)
  266. );
  267. }
  268. /**
  269. * {@inheritdoc}
  270. */
  271. public function createCommand($commandID, $arguments = array())
  272. {
  273. return $this->commands->create($commandID, $arguments);
  274. }
  275. /**
  276. * {@inheritdoc}
  277. */
  278. public function executeCommand(CommandInterface $command)
  279. {
  280. $response = $this->connection->executeCommand($command);
  281. if ($response instanceof ResponseInterface) {
  282. if ($response instanceof ErrorResponseInterface) {
  283. $response = $this->onErrorResponse($command, $response);
  284. }
  285. return $response;
  286. }
  287. return $command->parseResponse($response);
  288. }
  289. /**
  290. * Handles -ERR responses returned by Redis.
  291. *
  292. * @param CommandInterface $command Redis command that generated the error.
  293. * @param ErrorResponseInterface $response Instance of the error response.
  294. *
  295. * @throws ServerException
  296. *
  297. * @return mixed
  298. */
  299. protected function onErrorResponse(CommandInterface $command, ErrorResponseInterface $response)
  300. {
  301. if ($command instanceof ScriptCommand && $response->getErrorType() === 'NOSCRIPT') {
  302. $response = $this->executeCommand($command->getEvalCommand());
  303. if (!$response instanceof ResponseInterface) {
  304. $response = $command->parseResponse($response);
  305. }
  306. return $response;
  307. }
  308. if ($this->options->exceptions) {
  309. throw new ServerException($response->getMessage());
  310. }
  311. return $response;
  312. }
  313. /**
  314. * Executes the specified initializer method on `$this` by adjusting the
  315. * actual invokation depending on the arity (0, 1 or 2 arguments). This is
  316. * simply an utility method to create Redis contexts instances since they
  317. * follow a common initialization path.
  318. *
  319. * @param string $initializer Method name.
  320. * @param array $argv Arguments for the method.
  321. *
  322. * @return mixed
  323. */
  324. private function sharedContextFactory($initializer, $argv = null)
  325. {
  326. switch (count($argv)) {
  327. case 0:
  328. return $this->$initializer();
  329. case 1:
  330. return is_array($argv[0])
  331. ? $this->$initializer($argv[0])
  332. : $this->$initializer(null, $argv[0]);
  333. case 2:
  334. list($arg0, $arg1) = $argv;
  335. return $this->$initializer($arg0, $arg1);
  336. // @codeCoverageIgnoreStart
  337. default:
  338. return $this->$initializer($this, $argv);
  339. }
  340. // @codeCoverageIgnoreEnd
  341. }
  342. /**
  343. * Creates a new pipeline context and returns it, or returns the results of
  344. * a pipeline executed inside the optionally provided callable object.
  345. *
  346. * @param mixed ... Array of options, a callable for execution, or both.
  347. *
  348. * @return Pipeline|array
  349. */
  350. public function pipeline(/* arguments */)
  351. {
  352. return $this->sharedContextFactory('createPipeline', func_get_args());
  353. }
  354. /**
  355. * Actual pipeline context initializer method.
  356. *
  357. * @param array $options Options for the context.
  358. * @param mixed $callable Optional callable used to execute the context.
  359. *
  360. * @return Pipeline|array
  361. */
  362. protected function createPipeline(array $options = null, $callable = null)
  363. {
  364. if (isset($options['atomic']) && $options['atomic']) {
  365. $class = 'Predis\Pipeline\Atomic';
  366. } elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) {
  367. $class = 'Predis\Pipeline\FireAndForget';
  368. } else {
  369. $class = 'Predis\Pipeline\Pipeline';
  370. }
  371. /*
  372. * @var ClientContextInterface
  373. */
  374. $pipeline = new $class($this);
  375. if (isset($callable)) {
  376. return $pipeline->execute($callable);
  377. }
  378. return $pipeline;
  379. }
  380. /**
  381. * Creates a new transaction context and returns it, or returns the results
  382. * of a transaction executed inside the optionally provided callable object.
  383. *
  384. * @param mixed ... Array of options, a callable for execution, or both.
  385. *
  386. * @return MultiExecTransaction|array
  387. */
  388. public function transaction(/* arguments */)
  389. {
  390. return $this->sharedContextFactory('createTransaction', func_get_args());
  391. }
  392. /**
  393. * Actual transaction context initializer method.
  394. *
  395. * @param array $options Options for the context.
  396. * @param mixed $callable Optional callable used to execute the context.
  397. *
  398. * @return MultiExecTransaction|array
  399. */
  400. protected function createTransaction(array $options = null, $callable = null)
  401. {
  402. $transaction = new MultiExecTransaction($this, $options);
  403. if (isset($callable)) {
  404. return $transaction->execute($callable);
  405. }
  406. return $transaction;
  407. }
  408. /**
  409. * Creates a new publish/subscribe context and returns it, or starts its loop
  410. * inside the optionally provided callable object.
  411. *
  412. * @param mixed ... Array of options, a callable for execution, or both.
  413. *
  414. * @return PubSubConsumer|null
  415. */
  416. public function pubSubLoop(/* arguments */)
  417. {
  418. return $this->sharedContextFactory('createPubSub', func_get_args());
  419. }
  420. /**
  421. * Actual publish/subscribe context initializer method.
  422. *
  423. * @param array $options Options for the context.
  424. * @param mixed $callable Optional callable used to execute the context.
  425. *
  426. * @return PubSubConsumer|null
  427. */
  428. protected function createPubSub(array $options = null, $callable = null)
  429. {
  430. $pubsub = new PubSubConsumer($this, $options);
  431. if (!isset($callable)) {
  432. return $pubsub;
  433. }
  434. foreach ($pubsub as $message) {
  435. if (call_user_func($callable, $pubsub, $message) === false) {
  436. $pubsub->stop();
  437. }
  438. }
  439. }
  440. /**
  441. * Creates a new monitor consumer and returns it.
  442. *
  443. * @return MonitorConsumer
  444. */
  445. public function monitor()
  446. {
  447. return new MonitorConsumer($this);
  448. }
  449. /**
  450. * {@inheritdoc}
  451. */
  452. #[\ReturnTypeWillChange]
  453. public function getIterator()
  454. {
  455. $clients = array();
  456. $connection = $this->getConnection();
  457. if (!$connection instanceof \Traversable) {
  458. return new \ArrayIterator(array(
  459. (string) $connection => new static($connection, $this->getOptions())
  460. ));
  461. }
  462. foreach ($connection as $node) {
  463. $clients[(string) $node] = new static($node, $this->getOptions());
  464. }
  465. return new \ArrayIterator($clients);
  466. }
  467. }