Client.php 16 KB

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