PredisCluster.php 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235
  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\Connection\Aggregate;
  11. use Predis\Cluster\PredisStrategy;
  12. use Predis\Cluster\StrategyInterface;
  13. use Predis\Command\CommandInterface;
  14. use Predis\Connection\NodeConnectionInterface;
  15. use Predis\NotSupportedException;
  16. /**
  17. * Abstraction for a cluster of aggregate connections to various Redis servers
  18. * implementing client-side sharding based on pluggable distribution strategies.
  19. *
  20. * @author Daniele Alessandri <suppakilla@gmail.com>
  21. *
  22. * @todo Add the ability to remove connections from pool.
  23. */
  24. class PredisCluster implements ClusterInterface, \IteratorAggregate, \Countable
  25. {
  26. private $pool;
  27. private $strategy;
  28. private $distributor;
  29. /**
  30. * @param StrategyInterface $strategy Optional cluster strategy.
  31. */
  32. public function __construct(StrategyInterface $strategy = null)
  33. {
  34. $this->pool = array();
  35. $this->strategy = $strategy ?: new PredisStrategy();
  36. $this->distributor = $this->strategy->getDistributor();
  37. }
  38. /**
  39. * {@inheritdoc}
  40. */
  41. public function isConnected()
  42. {
  43. foreach ($this->pool as $connection) {
  44. if ($connection->isConnected()) {
  45. return true;
  46. }
  47. }
  48. return false;
  49. }
  50. /**
  51. * {@inheritdoc}
  52. */
  53. public function connect()
  54. {
  55. foreach ($this->pool as $connection) {
  56. $connection->connect();
  57. }
  58. }
  59. /**
  60. * {@inheritdoc}
  61. */
  62. public function disconnect()
  63. {
  64. foreach ($this->pool as $connection) {
  65. $connection->disconnect();
  66. }
  67. }
  68. /**
  69. * {@inheritdoc}
  70. */
  71. public function add(NodeConnectionInterface $connection)
  72. {
  73. $parameters = $connection->getParameters();
  74. if (isset($parameters->alias)) {
  75. $this->pool[$parameters->alias] = $connection;
  76. } else {
  77. $this->pool[] = $connection;
  78. }
  79. $weight = isset($parameters->weight) ? $parameters->weight : null;
  80. $this->distributor->add($connection, $weight);
  81. }
  82. /**
  83. * {@inheritdoc}
  84. */
  85. public function remove(NodeConnectionInterface $connection)
  86. {
  87. if (($id = array_search($connection, $this->pool, true)) !== false) {
  88. unset($this->pool[$id]);
  89. $this->distributor->remove($connection);
  90. return true;
  91. }
  92. return false;
  93. }
  94. /**
  95. * Removes a connection instance using its alias or index.
  96. *
  97. * @param string $connectionID Alias or index of a connection.
  98. *
  99. * @return bool Returns true if the connection was in the pool.
  100. */
  101. public function removeById($connectionID)
  102. {
  103. if ($connection = $this->getConnectionById($connectionID)) {
  104. return $this->remove($connection);
  105. }
  106. return false;
  107. }
  108. /**
  109. * {@inheritdoc}
  110. */
  111. public function getConnection(CommandInterface $command)
  112. {
  113. $slot = $this->strategy->getSlot($command);
  114. if (!isset($slot)) {
  115. throw new NotSupportedException(
  116. "Cannot use '{$command->getId()}' over clusters of connections."
  117. );
  118. }
  119. $node = $this->distributor->getBySlot($slot);
  120. return $node;
  121. }
  122. /**
  123. * {@inheritdoc}
  124. */
  125. public function getConnectionById($connectionID)
  126. {
  127. return isset($this->pool[$connectionID]) ? $this->pool[$connectionID] : null;
  128. }
  129. /**
  130. * Retrieves a connection instance from the cluster using a key.
  131. *
  132. * @param string $key Key string.
  133. *
  134. * @return NodeConnectionInterface
  135. */
  136. public function getConnectionByKey($key)
  137. {
  138. $hash = $this->strategy->getSlotByKey($key);
  139. $node = $this->distributor->getBySlot($hash);
  140. return $node;
  141. }
  142. /**
  143. * Returns the underlying command hash strategy used to hash commands by
  144. * using keys found in their arguments.
  145. *
  146. * @return StrategyInterface
  147. */
  148. public function getClusterStrategy()
  149. {
  150. return $this->strategy;
  151. }
  152. /**
  153. * {@inheritdoc}
  154. */
  155. public function count()
  156. {
  157. return count($this->pool);
  158. }
  159. /**
  160. * {@inheritdoc}
  161. */
  162. public function getIterator()
  163. {
  164. return new \ArrayIterator($this->pool);
  165. }
  166. /**
  167. * {@inheritdoc}
  168. */
  169. public function writeRequest(CommandInterface $command)
  170. {
  171. $this->getConnection($command)->writeRequest($command);
  172. }
  173. /**
  174. * {@inheritdoc}
  175. */
  176. public function readResponse(CommandInterface $command)
  177. {
  178. return $this->getConnection($command)->readResponse($command);
  179. }
  180. /**
  181. * {@inheritdoc}
  182. */
  183. public function executeCommand(CommandInterface $command)
  184. {
  185. return $this->getConnection($command)->executeCommand($command);
  186. }
  187. /**
  188. * Executes the specified Redis command on all the nodes of a cluster.
  189. *
  190. * @param CommandInterface $command A Redis command.
  191. *
  192. * @return array
  193. */
  194. public function executeCommandOnNodes(CommandInterface $command)
  195. {
  196. $responses = array();
  197. foreach ($this->pool as $connection) {
  198. $responses[] = $connection->executeCommand($command);
  199. }
  200. return $responses;
  201. }
  202. }