Client.php 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547
  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\AggregateConnectionInterface;
  17. use Predis\Connection\ConnectionInterface;
  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 = '1.1.2-dev';
  40. protected $connection;
  41. protected $options;
  42. private $profile;
  43. /**
  44. * @param mixed $parameters Connection parameters for one or more servers.
  45. * @param mixed $options Options to configure some behaviours of the client.
  46. */
  47. public function __construct($parameters = null, $options = null)
  48. {
  49. $this->options = $this->createOptions($options ?: array());
  50. $this->connection = $this->createConnection($parameters ?: array());
  51. $this->profile = $this->options->profile;
  52. }
  53. /**
  54. * Creates a new instance of Predis\Configuration\Options from different
  55. * types of arguments or simply returns the passed argument if it is an
  56. * instance of Predis\Configuration\OptionsInterface.
  57. *
  58. * @param mixed $options Client options.
  59. *
  60. * @throws \InvalidArgumentException
  61. *
  62. * @return OptionsInterface
  63. */
  64. protected function createOptions($options)
  65. {
  66. if (is_array($options)) {
  67. return new Options($options);
  68. }
  69. if ($options instanceof OptionsInterface) {
  70. return $options;
  71. }
  72. throw new \InvalidArgumentException('Invalid type for client options.');
  73. }
  74. /**
  75. * Creates single or aggregate connections from different types of arguments
  76. * (string, array) or returns the passed argument if it is an instance of a
  77. * class implementing Predis\Connection\ConnectionInterface.
  78. *
  79. * Accepted types for connection parameters are:
  80. *
  81. * - Instance of Predis\Connection\ConnectionInterface.
  82. * - Instance of Predis\Connection\ParametersInterface.
  83. * - Array
  84. * - String
  85. * - Callable
  86. *
  87. * @param mixed $parameters Connection parameters or connection instance.
  88. *
  89. * @throws \InvalidArgumentException
  90. *
  91. * @return ConnectionInterface
  92. */
  93. protected function createConnection($parameters)
  94. {
  95. if ($parameters instanceof ConnectionInterface) {
  96. return $parameters;
  97. }
  98. if ($parameters instanceof ParametersInterface || is_string($parameters)) {
  99. return $this->options->connections->create($parameters);
  100. }
  101. if (is_array($parameters)) {
  102. if (!isset($parameters[0])) {
  103. return $this->options->connections->create($parameters);
  104. }
  105. $options = $this->options;
  106. if ($options->defined('aggregate')) {
  107. $initializer = $this->getConnectionInitializerWrapper($options->aggregate);
  108. $connection = $initializer($parameters, $options);
  109. } elseif ($options->defined('replication')) {
  110. $replication = $options->replication;
  111. if ($replication instanceof AggregateConnectionInterface) {
  112. $connection = $replication;
  113. $options->connections->aggregate($connection, $parameters);
  114. } else {
  115. $initializer = $this->getConnectionInitializerWrapper($replication);
  116. $connection = $initializer($parameters, $options);
  117. }
  118. } else {
  119. $connection = $options->cluster;
  120. $options->connections->aggregate($connection, $parameters);
  121. }
  122. return $connection;
  123. }
  124. if (is_callable($parameters)) {
  125. $initializer = $this->getConnectionInitializerWrapper($parameters);
  126. $connection = $initializer($this->options);
  127. return $connection;
  128. }
  129. throw new \InvalidArgumentException('Invalid type for connection parameters.');
  130. }
  131. /**
  132. * Wraps a callable to make sure that its returned value represents a valid
  133. * connection type.
  134. *
  135. * @param mixed $callable
  136. *
  137. * @return \Closure
  138. */
  139. protected function getConnectionInitializerWrapper($callable)
  140. {
  141. return function () use ($callable) {
  142. $connection = call_user_func_array($callable, func_get_args());
  143. if (!$connection instanceof ConnectionInterface) {
  144. throw new \UnexpectedValueException(
  145. 'The callable connection initializer returned an invalid type.'
  146. );
  147. }
  148. return $connection;
  149. };
  150. }
  151. /**
  152. * {@inheritdoc}
  153. */
  154. public function getProfile()
  155. {
  156. return $this->profile;
  157. }
  158. /**
  159. * {@inheritdoc}
  160. */
  161. public function getOptions()
  162. {
  163. return $this->options;
  164. }
  165. /**
  166. * Creates a new client instance for the specified connection ID or alias,
  167. * only when working with an aggregate connection (cluster, replication).
  168. * The new client instances uses the same options of the original one.
  169. *
  170. * @param string $connectionID Identifier of a connection.
  171. *
  172. * @throws \InvalidArgumentException
  173. *
  174. * @return Client
  175. */
  176. public function getClientFor($connectionID)
  177. {
  178. if (!$connection = $this->getConnectionById($connectionID)) {
  179. throw new \InvalidArgumentException("Invalid connection ID: $connectionID.");
  180. }
  181. return new static($connection, $this->options);
  182. }
  183. /**
  184. * Opens the underlying connection and connects to the server.
  185. */
  186. public function connect()
  187. {
  188. $this->connection->connect();
  189. }
  190. /**
  191. * Closes the underlying connection and disconnects from the server.
  192. */
  193. public function disconnect()
  194. {
  195. $this->connection->disconnect();
  196. }
  197. /**
  198. * Closes the underlying connection and disconnects from the server.
  199. *
  200. * This is the same as `Client::disconnect()` as it does not actually send
  201. * the `QUIT` command to Redis, but simply closes the connection.
  202. */
  203. public function quit()
  204. {
  205. $this->disconnect();
  206. }
  207. /**
  208. * Returns the current state of the underlying connection.
  209. *
  210. * @return bool
  211. */
  212. public function isConnected()
  213. {
  214. return $this->connection->isConnected();
  215. }
  216. /**
  217. * {@inheritdoc}
  218. */
  219. public function getConnection()
  220. {
  221. return $this->connection;
  222. }
  223. /**
  224. * Retrieves the specified connection from the aggregate connection when the
  225. * client is in cluster or replication mode.
  226. *
  227. * @param string $connectionID Index or alias of the single connection.
  228. *
  229. * @throws NotSupportedException
  230. *
  231. * @return Connection\NodeConnectionInterface
  232. */
  233. public function getConnectionById($connectionID)
  234. {
  235. if (!$this->connection instanceof AggregateConnectionInterface) {
  236. throw new NotSupportedException(
  237. 'Retrieving connections by ID is supported only by aggregate connections.'
  238. );
  239. }
  240. return $this->connection->getConnectionById($connectionID);
  241. }
  242. /**
  243. * Executes a command without filtering its arguments, parsing the response,
  244. * applying any prefix to keys or throwing exceptions on Redis errors even
  245. * regardless of client options.
  246. *
  247. * It is possible to identify Redis error responses from normal responses
  248. * using the second optional argument which is populated by reference.
  249. *
  250. * @param array $arguments Command arguments as defined by the command signature.
  251. * @param bool $error Set to TRUE when Redis returned an error response.
  252. *
  253. * @return mixed
  254. */
  255. public function executeRaw(array $arguments, &$error = null)
  256. {
  257. $error = false;
  258. $response = $this->connection->executeCommand(
  259. new RawCommand($arguments)
  260. );
  261. if ($response instanceof ResponseInterface) {
  262. if ($response instanceof ErrorResponseInterface) {
  263. $error = true;
  264. }
  265. return (string) $response;
  266. }
  267. return $response;
  268. }
  269. /**
  270. * {@inheritdoc}
  271. */
  272. public function __call($commandID, $arguments)
  273. {
  274. return $this->executeCommand(
  275. $this->createCommand($commandID, $arguments)
  276. );
  277. }
  278. /**
  279. * {@inheritdoc}
  280. */
  281. public function createCommand($commandID, $arguments = array())
  282. {
  283. return $this->profile->createCommand($commandID, $arguments);
  284. }
  285. /**
  286. * {@inheritdoc}
  287. */
  288. public function executeCommand(CommandInterface $command)
  289. {
  290. $response = $this->connection->executeCommand($command);
  291. if ($response instanceof ResponseInterface) {
  292. if ($response instanceof ErrorResponseInterface) {
  293. $response = $this->onErrorResponse($command, $response);
  294. }
  295. return $response;
  296. }
  297. return $command->parseResponse($response);
  298. }
  299. /**
  300. * Handles -ERR responses returned by Redis.
  301. *
  302. * @param CommandInterface $command Redis command that generated the error.
  303. * @param ErrorResponseInterface $response Instance of the error response.
  304. *
  305. * @throws ServerException
  306. *
  307. * @return mixed
  308. */
  309. protected function onErrorResponse(CommandInterface $command, ErrorResponseInterface $response)
  310. {
  311. if ($command instanceof ScriptCommand && $response->getErrorType() === 'NOSCRIPT') {
  312. $eval = $this->createCommand('EVAL');
  313. $eval->setRawArguments($command->getEvalArguments());
  314. $response = $this->executeCommand($eval);
  315. if (!$response instanceof ResponseInterface) {
  316. $response = $command->parseResponse($response);
  317. }
  318. return $response;
  319. }
  320. if ($this->options->exceptions) {
  321. throw new ServerException($response->getMessage());
  322. }
  323. return $response;
  324. }
  325. /**
  326. * Executes the specified initializer method on `$this` by adjusting the
  327. * actual invokation depending on the arity (0, 1 or 2 arguments). This is
  328. * simply an utility method to create Redis contexts instances since they
  329. * follow a common initialization path.
  330. *
  331. * @param string $initializer Method name.
  332. * @param array $argv Arguments for the method.
  333. *
  334. * @return mixed
  335. */
  336. private function sharedContextFactory($initializer, $argv = null)
  337. {
  338. switch (count($argv)) {
  339. case 0:
  340. return $this->$initializer();
  341. case 1:
  342. return is_array($argv[0])
  343. ? $this->$initializer($argv[0])
  344. : $this->$initializer(null, $argv[0]);
  345. case 2:
  346. list($arg0, $arg1) = $argv;
  347. return $this->$initializer($arg0, $arg1);
  348. default:
  349. return $this->$initializer($this, $argv);
  350. }
  351. }
  352. /**
  353. * Creates a new pipeline context and returns it, or returns the results of
  354. * a pipeline executed inside the optionally provided callable object.
  355. *
  356. * @param mixed ... Array of options, a callable for execution, or both.
  357. *
  358. * @return Pipeline|array
  359. */
  360. public function pipeline(/* arguments */)
  361. {
  362. return $this->sharedContextFactory('createPipeline', func_get_args());
  363. }
  364. /**
  365. * Actual pipeline context initializer method.
  366. *
  367. * @param array $options Options for the context.
  368. * @param mixed $callable Optional callable used to execute the context.
  369. *
  370. * @return Pipeline|array
  371. */
  372. protected function createPipeline(array $options = null, $callable = null)
  373. {
  374. if (isset($options['atomic']) && $options['atomic']) {
  375. $class = 'Predis\Pipeline\Atomic';
  376. } elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) {
  377. $class = 'Predis\Pipeline\FireAndForget';
  378. } else {
  379. $class = 'Predis\Pipeline\Pipeline';
  380. }
  381. /*
  382. * @var ClientContextInterface
  383. */
  384. $pipeline = new $class($this);
  385. if (isset($callable)) {
  386. return $pipeline->execute($callable);
  387. }
  388. return $pipeline;
  389. }
  390. /**
  391. * Creates a new transaction context and returns it, or returns the results
  392. * of a transaction executed inside the optionally provided callable object.
  393. *
  394. * @param mixed ... Array of options, a callable for execution, or both.
  395. *
  396. * @return MultiExecTransaction|array
  397. */
  398. public function transaction(/* arguments */)
  399. {
  400. return $this->sharedContextFactory('createTransaction', func_get_args());
  401. }
  402. /**
  403. * Actual transaction context initializer method.
  404. *
  405. * @param array $options Options for the context.
  406. * @param mixed $callable Optional callable used to execute the context.
  407. *
  408. * @return MultiExecTransaction|array
  409. */
  410. protected function createTransaction(array $options = null, $callable = null)
  411. {
  412. $transaction = new MultiExecTransaction($this, $options);
  413. if (isset($callable)) {
  414. return $transaction->execute($callable);
  415. }
  416. return $transaction;
  417. }
  418. /**
  419. * Creates a new publish/subscribe context and returns it, or starts its loop
  420. * inside the optionally provided callable object.
  421. *
  422. * @param mixed ... Array of options, a callable for execution, or both.
  423. *
  424. * @return PubSubConsumer|null
  425. */
  426. public function pubSubLoop(/* arguments */)
  427. {
  428. return $this->sharedContextFactory('createPubSub', func_get_args());
  429. }
  430. /**
  431. * Actual publish/subscribe context initializer method.
  432. *
  433. * @param array $options Options for the context.
  434. * @param mixed $callable Optional callable used to execute the context.
  435. *
  436. * @return PubSubConsumer|null
  437. */
  438. protected function createPubSub(array $options = null, $callable = null)
  439. {
  440. $pubsub = new PubSubConsumer($this, $options);
  441. if (!isset($callable)) {
  442. return $pubsub;
  443. }
  444. foreach ($pubsub as $message) {
  445. if (call_user_func($callable, $pubsub, $message) === false) {
  446. $pubsub->stop();
  447. }
  448. }
  449. }
  450. /**
  451. * Creates a new monitor consumer and returns it.
  452. *
  453. * @return MonitorConsumer
  454. */
  455. public function monitor()
  456. {
  457. return new MonitorConsumer($this);
  458. }
  459. /**
  460. * {@inheritdoc}
  461. */
  462. public function getIterator()
  463. {
  464. $clients = array();
  465. $connection = $this->getConnection();
  466. if (!$connection instanceof \Traversable) {
  467. throw new ClientException('The underlying connection is not traversable');
  468. }
  469. foreach ($connection as $node) {
  470. $clients[(string) $node] = new static($node, $this->getOptions());
  471. }
  472. return new \ArrayIterator($clients);
  473. }
  474. }