123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290 |
- <?php
- namespace Predis\PubSub;
- use Predis\Client;
- use Predis\Helpers;
- use Predis\ClientException;
- use Predis\NotSupportedException;
- class PubSubContext implements \Iterator
- {
- const SUBSCRIBE = 'subscribe';
- const UNSUBSCRIBE = 'unsubscribe';
- const PSUBSCRIBE = 'psubscribe';
- const PUNSUBSCRIBE = 'punsubscribe';
- const MESSAGE = 'message';
- const PMESSAGE = 'pmessage';
- const STATUS_VALID = 1;
- const STATUS_SUBSCRIBED = 2;
- const STATUS_PSUBSCRIBED = 4;
- private $client;
- private $position;
- private $options;
-
- public function __construct(Client $client, Array $options = null)
- {
- $this->checkCapabilities($client);
- $this->options = $options ?: array();
- $this->client = $client;
- $this->statusFlags = self::STATUS_VALID;
- $this->genericSubscribeInit('subscribe');
- $this->genericSubscribeInit('psubscribe');
- }
-
- public function __destruct()
- {
- $this->closeContext(true);
- }
-
- private function checkCapabilities(Client $client)
- {
- if (Helpers::isCluster($client->getConnection())) {
- throw new NotSupportedException('Cannot initialize a PUB/SUB context over a cluster of connections');
- }
- $commands = array('publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe');
- if ($client->getProfile()->supportsCommands($commands) === false) {
- throw new NotSupportedException('The current profile does not support PUB/SUB related commands');
- }
- }
-
- private function genericSubscribeInit($subscribeAction)
- {
- if (isset($this->options[$subscribeAction])) {
- $this->$subscribeAction($this->options[$subscribeAction]);
- }
- }
-
- private function isFlagSet($value)
- {
- return ($this->statusFlags & $value) === $value;
- }
-
- public function subscribe()
- {
- $this->writeCommand(self::SUBSCRIBE, func_get_args());
- $this->statusFlags |= self::STATUS_SUBSCRIBED;
- }
-
- public function unsubscribe()
- {
- $this->writeCommand(self::UNSUBSCRIBE, func_get_args());
- }
-
- public function psubscribe()
- {
- $this->writeCommand(self::PSUBSCRIBE, func_get_args());
- $this->statusFlags |= self::STATUS_PSUBSCRIBED;
- }
-
- public function punsubscribe()
- {
- $this->writeCommand(self::PUNSUBSCRIBE, func_get_args());
- }
-
- public function closeContext($force = false)
- {
- if (!$this->valid()) {
- return false;
- }
- if ($force) {
- $this->invalidate();
- $this->client->disconnect();
- }
- else {
- if ($this->isFlagSet(self::STATUS_SUBSCRIBED)) {
- $this->unsubscribe();
- }
- if ($this->isFlagSet(self::STATUS_PSUBSCRIBED)) {
- $this->punsubscribe();
- }
- }
- return !$force;
- }
-
- private function writeCommand($method, $arguments)
- {
- $arguments = Helpers::filterArrayArguments($arguments);
- $command = $this->client->createCommand($method, $arguments);
- $this->client->getConnection()->writeCommand($command);
- }
-
- public function rewind()
- {
-
- }
-
- public function current()
- {
- return $this->getValue();
- }
-
- public function key()
- {
- return $this->position;
- }
-
- public function next()
- {
- if ($this->valid()) {
- $this->position++;
- }
- return $this->position;
- }
-
- public function valid()
- {
- $isValid = $this->isFlagSet(self::STATUS_VALID);
- $subscriptionFlags = self::STATUS_SUBSCRIBED | self::STATUS_PSUBSCRIBED;
- $hasSubscriptions = ($this->statusFlags & $subscriptionFlags) > 0;
- return $isValid && $hasSubscriptions;
- }
-
- private function invalidate()
- {
- $this->statusFlags = 0;
- }
-
- private function getValue()
- {
- $response = $this->client->getConnection()->read();
- switch ($response[0]) {
- case self::SUBSCRIBE:
- case self::UNSUBSCRIBE:
- case self::PSUBSCRIBE:
- case self::PUNSUBSCRIBE:
- if ($response[2] === 0) {
- $this->invalidate();
- }
- case self::MESSAGE:
- return (object) array(
- 'kind' => $response[0],
- 'channel' => $response[1],
- 'payload' => $response[2],
- );
- case self::PMESSAGE:
- return (object) array(
- 'kind' => $response[0],
- 'pattern' => $response[1],
- 'channel' => $response[2],
- 'payload' => $response[3],
- );
- default:
- $message = "Received an unknown message type {$response[0]} inside of a pubsub context";
- throw new ClientException($message);
- }
- }
- }
|