Client.php 15 KB

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