From 7312dbe7fc817945b75e255722d3322aa32e5666 Mon Sep 17 00:00:00 2001 From: Daniele Alessandri Date: Thu, 13 May 2010 16:07:42 +0200 Subject: [PATCH 1/5] Implemented the new \Predis\PubSubContext class. --- lib/Predis.php | 120 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) diff --git a/lib/Predis.php b/lib/Predis.php index c1530095..f489205c 100644 --- a/lib/Predis.php +++ b/lib/Predis.php @@ -834,6 +834,126 @@ class MultiExecBlock { } } +class PubSubContext implements \Iterator { + const SUBSCRIBE = 'subscribe'; + const UNSUBSCRIBE = 'unsubscribe'; + const PSUBSCRIBE = 'psubscribe'; + const PUNSUBSCRIBE = 'punsubscribe'; + const MESSAGE = 'message'; + const PMESSAGE = 'pmessage'; + + private $_redisClient, $_subscriptions, $_isStillValid, $_position; + + public function __construct(Client $redisClient) { + $this->_redisClient = $redisClient; + $this->_isStillValid = true; + $this->_subscriptions = false; + } + + public function __destruct() { + if ($this->valid()) { + $this->_redisClient->unsubscribe(); + $this->_redisClient->punsubscribe(); + } + } + + public function subscribe(/* arguments */) { + $this->writeCommand(self::SUBSCRIBE, func_get_args()); + $this->_subscriptions = true; + } + + public function unsubscribe(/* arguments */) { + $this->writeCommand(self::UNSUBSCRIBE, func_get_args()); + } + + public function psubscribe(/* arguments */) { + $this->writeCommand(self::PSUBSCRIBE, func_get_args()); + $this->_subscriptions = true; + } + + public function punsubscribe(/* arguments */) { + $this->writeCommand(self::PUNSUBSCRIBE, func_get_args()); + } + + public function closeContext() { + if ($this->valid()) { + // TODO: as an optimization, we should not send both + // commands if one of them has not been issued. + $this->unsubscribe(); + $this->punsubscribe(); + } + } + + private function writeCommand($method, $arguments) { + if (count($arguments) === 1 && is_array($arguments[0])) { + $arguments = $arguments[0]; + } + $command = $this->_redisClient->createCommand($method, $arguments); + $this->_redisClient->getConnection()->writeCommand($command); + } + + public function rewind() { + // NOOP + } + + public function current() { + return $this->getValue(); + } + + public function key() { + return $this->_position; + } + + public function next() { + if ($this->_isStillValid) { + $this->_position++; + } + return $this->_position; + } + + public function valid() { + return $this->_subscriptions && $this->_isStillValid; + } + + private function invalidate() { + $this->_isStillValid = false; + $this->_subscriptions = false; + } + + private function getValue() { + $reader = $this->_redisClient->getResponseReader(); + $connection = $this->_redisClient->getConnection(); + $response = $reader->read($connection); + + 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: + throw new \Predis\ClientException( + "Received an unknown message type {$response[0]} inside of a pubsub context" + ); + } + } +} + /* ------------------------------------------------------------------------- */ class ConnectionParameters { From a0f441f21b1bb7dd5d37b5bfbc7dfaffe4d1991a Mon Sep 17 00:00:00 2001 From: Daniele Alessandri Date: Thu, 13 May 2010 16:10:18 +0200 Subject: [PATCH 2/5] Added an example to show how to use the \Predis\PubSubContext class. --- examples/PubSubContext.php | 49 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 examples/PubSubContext.php diff --git a/examples/PubSubContext.php b/examples/PubSubContext.php new file mode 100644 index 00000000..9e6d03a6 --- /dev/null +++ b/examples/PubSubContext.php @@ -0,0 +1,49 @@ +subscribe('control_channel'); +$pubsub->subscribe('notifications'); + +// Start processing the pubsup messages. Open a terminal and use redis-cli +// to push messages to the channels. Examples: +// ./redis-cli PUBLISH notifications "this is a test" +// ./redis-cli PUBLISH control_channel quit_loop +foreach ($pubsub as $i => $message) { + switch ($message->kind) { + case 'subscribe': + echo "Subscribed to {$message->channel}\n"; + break; + case 'message': + if ($message->channel == 'control_channel') { + if ($message->payload == 'quit_loop') { + echo "Aborting pubsub loop...\n"; + $pubsub->unsubscribe(); + } + else { + echo "Received an unregognized command: {$message->payload}.\n"; + } + } + else { + echo "Received the following message from {$message->channel}:\n", + " {$message->payload}\n\n"; + } + break; + } +} + +// Always unset the pubsub context instance when you are done! The +// class destructor will take care of cleanups and prevent protocol +// desynchronizations between the client and the server. +unset($pubsub); + +// Say goodbye :-) +$info = $redis->info(); +print_r("Goodbye from Redis v{$info['redis_version']}!\n"); +?> From f6b3912783bd292a6ffabed7f83acf1348fbdf24 Mon Sep 17 00:00:00 2001 From: Daniele Alessandri Date: Thu, 13 May 2010 16:11:40 +0200 Subject: [PATCH 3/5] Updated TODO. --- TODO | 2 -- 1 file changed, 2 deletions(-) diff --git a/TODO b/TODO index 06b91361..ce9cd448 100644 --- a/TODO +++ b/TODO @@ -5,7 +5,5 @@ full battery of tests targeting specific functions of this library is still missing. -* Add a PubSubBlock class. - * Missing tests for new commands: PubSub : PUBLISH, SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE From ea7e9a37c6766cbc1ca3795accbb40cf4ca243b3 Mon Sep 17 00:00:00 2001 From: Daniele Alessandri Date: Tue, 18 May 2010 11:54:59 +0200 Subject: [PATCH 4/5] Remove unused variable in the pubsub example. --- examples/PubSubContext.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/PubSubContext.php b/examples/PubSubContext.php index 9e6d03a6..7d7fe72a 100644 --- a/examples/PubSubContext.php +++ b/examples/PubSubContext.php @@ -15,7 +15,7 @@ $pubsub->subscribe('notifications'); // to push messages to the channels. Examples: // ./redis-cli PUBLISH notifications "this is a test" // ./redis-cli PUBLISH control_channel quit_loop -foreach ($pubsub as $i => $message) { +foreach ($pubsub as $message) { switch ($message->kind) { case 'subscribe': echo "Subscribed to {$message->channel}\n"; From 6e321a3530aa987844e8a2deefc85f6375ef3f17 Mon Sep 17 00:00:00 2001 From: Daniele Alessandri Date: Tue, 18 May 2010 12:23:53 +0200 Subject: [PATCH 5/5] Add the Predis\Client::pubSubContext() method to initialize a PubSub context from a client instance. --- examples/PubSubContext.php | 2 +- lib/Predis.php | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/examples/PubSubContext.php b/examples/PubSubContext.php index 7d7fe72a..38b8a2fb 100644 --- a/examples/PubSubContext.php +++ b/examples/PubSubContext.php @@ -5,7 +5,7 @@ require_once '../lib/Predis.php'; $redis = new \Predis\Client('redis://127.0.0.1:6379/?read_write_timeout=-1', 'dev'); // Initialize a new pubsub context -$pubsub = new \Predis\PubSubContext($redis); +$pubsub = $redis->pubSubContext(); // Subscribe to your channels $pubsub->subscribe('control_channel'); diff --git a/lib/Predis.php b/lib/Predis.php index f489205c..d8045b3c 100644 --- a/lib/Predis.php +++ b/lib/Predis.php @@ -216,6 +216,10 @@ class Client { $multiExec = new MultiExecBlock($this); return $multiExecBlock !== null ? $multiExec->execute($multiExecBlock) : $multiExec; } + + public function pubSubContext() { + return new PubSubContext($this); + } } /* ------------------------------------------------------------------------- */