Client.php 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212
  1. <?php
  2. namespace Predis;
  3. use Predis\Commands\ICommand;
  4. use Predis\Network\IConnection;
  5. use Predis\Network\IConnectionSingle;
  6. use Predis\Network\ConnectionCluster;
  7. use Predis\Profiles\ServerProfile;
  8. use Predis\Profiles\IServerProfile;
  9. class Client {
  10. const VERSION = '0.7.0-dev';
  11. private $_options, $_schemes, $_profile, $_connection;
  12. public function __construct($parameters = null, $options = null) {
  13. $options = $this->filterOptions($options ?: new ClientOptions());
  14. $this->_options = $options;
  15. $this->_profile = $options->profile;
  16. $this->_schemes = $options->connections;
  17. $this->_connection = $this->initializeConnection($parameters);
  18. }
  19. private function filterOptions($options) {
  20. if ($options instanceof ClientOptions) {
  21. return $options;
  22. }
  23. if (is_array($options)) {
  24. return new ClientOptions($options);
  25. }
  26. if ($options instanceof IServerProfile) {
  27. return new ClientOptions(array('profile' => $options));
  28. }
  29. if (is_string($options)) {
  30. return new ClientOptions(array('profile' => ServerProfile::get($options)));
  31. }
  32. throw new \InvalidArgumentException("Invalid type for client options");
  33. }
  34. private function initializeConnection($parameters = array()) {
  35. if ($parameters === null) {
  36. return $this->createConnection(array());
  37. }
  38. if ($parameters instanceof IConnection) {
  39. return $parameters;
  40. }
  41. if (is_array($parameters) && isset($parameters[0])) {
  42. $cluster = new ConnectionCluster($this->_options->key_distribution);
  43. foreach ($parameters as $single) {
  44. $cluster->add($single instanceof IConnectionSingle
  45. ? $single : $this->createConnection($single)
  46. );
  47. }
  48. return $cluster;
  49. }
  50. return $this->createConnection($parameters);
  51. }
  52. private function createConnection($parameters) {
  53. $connection = $this->_schemes->newConnection($parameters);
  54. $this->pushInitCommands($connection);
  55. return $connection;
  56. }
  57. private function pushInitCommands(IConnectionSingle $connection) {
  58. $params = $connection->getParameters();
  59. if (isset($params->password)) {
  60. $connection->pushInitCommand($this->createCommand(
  61. 'auth', array($params->password)
  62. ));
  63. }
  64. if (isset($params->database)) {
  65. $connection->pushInitCommand($this->createCommand(
  66. 'select', array($params->database)
  67. ));
  68. }
  69. }
  70. public function getProfile() {
  71. return $this->_profile;
  72. }
  73. public function getOptions() {
  74. return $this->_options;
  75. }
  76. public function getSchemes() {
  77. return $this->_schemes;
  78. }
  79. public function getClientFor($connectionAlias) {
  80. if (!Utils::isCluster($this->_connection)) {
  81. throw new ClientException(
  82. 'This method is supported only when the client is connected to a cluster of connections'
  83. );
  84. }
  85. $connection = $this->_connection->getConnectionById($connectionAlias);
  86. if ($connection === null) {
  87. throw new \InvalidArgumentException(
  88. "Invalid connection alias: '$connectionAlias'"
  89. );
  90. }
  91. return new Client($connection, $this->_options);
  92. }
  93. public function connect() {
  94. $this->_connection->connect();
  95. }
  96. public function disconnect() {
  97. $this->_connection->disconnect();
  98. }
  99. public function quit() {
  100. $this->disconnect();
  101. }
  102. public function isConnected() {
  103. return $this->_connection->isConnected();
  104. }
  105. public function getConnection($id = null) {
  106. if ($id === null) {
  107. return $this->_connection;
  108. }
  109. $connection = $this->_connection;
  110. $isCluster = Utils::isCluster($connection);
  111. return $isCluster ? $connection->getConnectionById($id) : $connection;
  112. }
  113. public function __call($method, $arguments) {
  114. $command = $this->_profile->createCommand($method, $arguments);
  115. return $this->_connection->executeCommand($command);
  116. }
  117. public function createCommand($method, $arguments = array()) {
  118. return $this->_profile->createCommand($method, $arguments);
  119. }
  120. public function executeCommand(ICommand $command) {
  121. return $this->_connection->executeCommand($command);
  122. }
  123. public function executeCommandOnShards(ICommand $command) {
  124. if (Utils::isCluster($this->_connection)) {
  125. $replies = array();
  126. foreach ($this->_connection as $connection) {
  127. $replies[] = $connection->executeCommand($command);
  128. }
  129. return $replies;
  130. }
  131. return array($this->_connection->executeCommand($command));
  132. }
  133. private function sharedInitializer($argv, $initializer) {
  134. $argc = count($argv);
  135. if ($argc === 0) {
  136. return $this->$initializer();
  137. }
  138. else if ($argc === 1) {
  139. list($arg0) = $argv;
  140. return is_array($arg0) ? $this->$initializer($arg0) : $this->$initializer(null, $arg0);
  141. }
  142. else if ($argc === 2) {
  143. list($arg0, $arg1) = $argv;
  144. return $this->$initializer($arg0, $arg1);
  145. }
  146. return $this->$initializer($this, $arguments);
  147. }
  148. public function pipeline(/* arguments */) {
  149. return $this->sharedInitializer(func_get_args(), 'initPipeline');
  150. }
  151. private function initPipeline(Array $options = null, $pipelineBlock = null) {
  152. $pipeline = null;
  153. if (isset($options)) {
  154. if (isset($options['safe']) && $options['safe'] == true) {
  155. $connection = $this->_connection;
  156. $pipeline = new CommandPipeline($this,
  157. Utils::isCluster($connection)
  158. ? new Pipeline\SafeClusterExecutor($connection)
  159. : new Pipeline\SafeExecutor($connection)
  160. );
  161. }
  162. else {
  163. $pipeline = new CommandPipeline($this);
  164. }
  165. }
  166. return $this->pipelineExecute(
  167. $pipeline ?: new CommandPipeline($this), $pipelineBlock
  168. );
  169. }
  170. private function pipelineExecute(CommandPipeline $pipeline, $block) {
  171. return $block !== null ? $pipeline->execute($block) : $pipeline;
  172. }
  173. public function multiExec(/* arguments */) {
  174. return $this->sharedInitializer(func_get_args(), 'initMultiExec');
  175. }
  176. private function initMultiExec(Array $options = null, $transBlock = null) {
  177. $multi = isset($options) ? new MultiExecContext($this, $options) : new MultiExecContext($this);
  178. return $transBlock !== null ? $multi->execute($transBlock) : $multi;
  179. }
  180. public function pubSubContext(Array $options = null) {
  181. return new PubSubContext($this, $options);
  182. }
  183. }