PubSubContext.php 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180
  1. <?php
  2. namespace Predis;
  3. class PubSubContext implements \Iterator
  4. {
  5. const SUBSCRIBE = 'subscribe';
  6. const UNSUBSCRIBE = 'unsubscribe';
  7. const PSUBSCRIBE = 'psubscribe';
  8. const PUNSUBSCRIBE = 'punsubscribe';
  9. const MESSAGE = 'message';
  10. const PMESSAGE = 'pmessage';
  11. const STATUS_VALID = 0x0001;
  12. const STATUS_SUBSCRIBED = 0x0010;
  13. const STATUS_PSUBSCRIBED = 0x0100;
  14. private $_client;
  15. private $_position;
  16. private $_options;
  17. public function __construct(Client $client, Array $options = null)
  18. {
  19. $this->checkCapabilities($client);
  20. $this->_options = $options ?: array();
  21. $this->_client = $client;
  22. $this->_statusFlags = self::STATUS_VALID;
  23. $this->genericSubscribeInit('subscribe');
  24. $this->genericSubscribeInit('psubscribe');
  25. }
  26. public function __destruct()
  27. {
  28. $this->closeContext();
  29. }
  30. private function checkCapabilities(Client $client)
  31. {
  32. if (Helpers::isCluster($client->getConnection())) {
  33. throw new ClientException(
  34. 'Cannot initialize a PUB/SUB context over a cluster of connections'
  35. );
  36. }
  37. $commands = array('publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe');
  38. if ($client->getProfile()->supportsCommands($commands) === false) {
  39. throw new ClientException(
  40. 'The current profile does not support PUB/SUB related commands'
  41. );
  42. }
  43. }
  44. private function genericSubscribeInit($subscribeAction)
  45. {
  46. if (isset($this->_options[$subscribeAction])) {
  47. $this->$subscribeAction($this->_options[$subscribeAction]);
  48. }
  49. }
  50. private function isFlagSet($value)
  51. {
  52. return ($this->_statusFlags & $value) === $value;
  53. }
  54. public function subscribe(/* arguments */)
  55. {
  56. $this->writeCommand(self::SUBSCRIBE, func_get_args());
  57. $this->_statusFlags |= self::STATUS_SUBSCRIBED;
  58. }
  59. public function unsubscribe(/* arguments */)
  60. {
  61. $this->writeCommand(self::UNSUBSCRIBE, func_get_args());
  62. }
  63. public function psubscribe(/* arguments */)
  64. {
  65. $this->writeCommand(self::PSUBSCRIBE, func_get_args());
  66. $this->_statusFlags |= self::STATUS_PSUBSCRIBED;
  67. }
  68. public function punsubscribe(/* arguments */)
  69. {
  70. $this->writeCommand(self::PUNSUBSCRIBE, func_get_args());
  71. }
  72. public function closeContext()
  73. {
  74. if ($this->valid()) {
  75. if ($this->isFlagSet(self::STATUS_SUBSCRIBED)) {
  76. $this->unsubscribe();
  77. }
  78. if ($this->isFlagSet(self::STATUS_PSUBSCRIBED)) {
  79. $this->punsubscribe();
  80. }
  81. }
  82. }
  83. private function writeCommand($method, $arguments)
  84. {
  85. $arguments = Helpers::filterArrayArguments($arguments);
  86. $command = $this->_client->createCommand($method, $arguments);
  87. $this->_client->getConnection()->writeCommand($command);
  88. }
  89. public function rewind()
  90. {
  91. // NOOP
  92. }
  93. public function current()
  94. {
  95. return $this->getValue();
  96. }
  97. public function key()
  98. {
  99. return $this->_position;
  100. }
  101. public function next()
  102. {
  103. if ($this->isFlagSet(self::STATUS_VALID)) {
  104. $this->_position++;
  105. }
  106. return $this->_position;
  107. }
  108. public function valid()
  109. {
  110. $isValid = $this->isFlagSet(self::STATUS_VALID);
  111. $subscriptionFlags = self::STATUS_SUBSCRIBED | self::STATUS_PSUBSCRIBED;
  112. $hasSubscriptions = ($this->_statusFlags & $subscriptionFlags) > 0;
  113. return $isValid && $hasSubscriptions;
  114. }
  115. private function invalidate()
  116. {
  117. $this->_statusFlags = 0x0000;
  118. }
  119. private function getValue()
  120. {
  121. $response = $this->_client->getConnection()->read();
  122. switch ($response[0]) {
  123. case self::SUBSCRIBE:
  124. case self::UNSUBSCRIBE:
  125. case self::PSUBSCRIBE:
  126. case self::PUNSUBSCRIBE:
  127. if ($response[2] === 0) {
  128. $this->invalidate();
  129. }
  130. case self::MESSAGE:
  131. return (object) array(
  132. 'kind' => $response[0],
  133. 'channel' => $response[1],
  134. 'payload' => $response[2],
  135. );
  136. case self::PMESSAGE:
  137. return (object) array(
  138. 'kind' => $response[0],
  139. 'pattern' => $response[1],
  140. 'channel' => $response[2],
  141. 'payload' => $response[3],
  142. );
  143. default:
  144. throw new ClientException(
  145. "Received an unknown message type {$response[0]} inside of a pubsub context"
  146. );
  147. }
  148. }
  149. }