ConnectionCluster.php 2.3 KB

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