PredisCluster.php 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. <?php
  2. namespace Predis\Network;
  3. use Predis\Helpers;
  4. use Predis\ClientException;
  5. use Predis\Commands\ICommand;
  6. use Predis\Distribution\IDistributionStrategy;
  7. use Predis\Distribution\HashRing;
  8. class PredisCluster implements IConnectionCluster, \IteratorAggregate
  9. {
  10. private $_pool;
  11. private $_distributor;
  12. public function __construct(IDistributionStrategy $distributor = null)
  13. {
  14. $this->_pool = array();
  15. $this->_distributor = $distributor ?: new HashRing();
  16. }
  17. public function isConnected()
  18. {
  19. foreach ($this->_pool as $connection) {
  20. if ($connection->isConnected()) {
  21. return true;
  22. }
  23. }
  24. return false;
  25. }
  26. public function connect()
  27. {
  28. foreach ($this->_pool as $connection) {
  29. $connection->connect();
  30. }
  31. }
  32. public function disconnect()
  33. {
  34. foreach ($this->_pool as $connection) {
  35. $connection->disconnect();
  36. }
  37. }
  38. public function add(IConnectionSingle $connection)
  39. {
  40. $parameters = $connection->getParameters();
  41. if (isset($parameters->alias)) {
  42. $this->_pool[$parameters->alias] = $connection;
  43. }
  44. else {
  45. $this->_pool[] = $connection;
  46. }
  47. $this->_distributor->add($connection, $parameters->weight);
  48. }
  49. public function getConnection(ICommand $command)
  50. {
  51. $cmdHash = $command->getHash($this->_distributor);
  52. if (isset($cmdHash)) {
  53. return $this->_distributor->get($cmdHash);
  54. }
  55. throw new ClientException(
  56. sprintf("Cannot send '%s' commands to a cluster of connections", $command->getId())
  57. );
  58. }
  59. public function getConnectionById($id = null)
  60. {
  61. $alias = $id ?: 0;
  62. return isset($this->_pool[$alias]) ? $this->_pool[$alias] : null;
  63. }
  64. public function getConnectionByKey($key)
  65. {
  66. $hashablePart = Helpers::getKeyHashablePart($key);
  67. $keyHash = $this->_distributor->generateKey($hashablePart);
  68. return $this->_distributor->get($keyHash);
  69. }
  70. public function getIterator()
  71. {
  72. return new \ArrayIterator($this->_pool);
  73. }
  74. public function writeCommand(ICommand $command)
  75. {
  76. $this->getConnection($command)->writeCommand($command);
  77. }
  78. public function readResponse(ICommand $command)
  79. {
  80. return $this->getConnection($command)->readResponse($command);
  81. }
  82. public function executeCommand(ICommand $command)
  83. {
  84. return $this->getConnection($command)->executeCommand($command);
  85. }
  86. }