Client.php 13 KB

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