diff --git a/lib/Predis/PubSub/DispatcherLoop.php b/lib/Predis/PubSub/DispatcherLoop.php index e2455a00..fa3ba7d2 100644 --- a/lib/Predis/PubSub/DispatcherLoop.php +++ b/lib/Predis/PubSub/DispatcherLoop.php @@ -21,11 +21,11 @@ use Predis\ClientInterface; */ class DispatcherLoop { - private $client; private $pubSubContext; - private $callbacks; - private $defaultCallback; - private $subscriptionCallback; + + protected $callbacks; + protected $defaultCallback; + protected $subscriptionCallback; /** * @param ClientInterface Client instance used by the context. @@ -33,7 +33,6 @@ class DispatcherLoop public function __construct(ClientInterface $client) { $this->callbacks = array(); - $this->client = $client; $this->pubSubContext = $client->pubSub(); } @@ -96,8 +95,10 @@ class DispatcherLoop */ public function attachCallback($channel, $callback) { + $callbackName = $this->getPrefixKeys() . $channel; + $this->validateCallback($callback); - $this->callbacks[$channel] = $callback; + $this->callbacks[$callbackName] = $callback; $this->pubSubContext->subscribe($channel); } @@ -108,8 +109,10 @@ class DispatcherLoop */ public function detachCallback($channel) { - if (isset($this->callbacks[$channel])) { - unset($this->callbacks[$channel]); + $callbackName = $this->getPrefixKeys() . $channel; + + if (isset($this->callbacks[$callbackName])) { + unset($this->callbacks[$callbackName]); $this->pubSubContext->unsubscribe($channel); } } @@ -148,4 +151,20 @@ class DispatcherLoop { $this->pubSubContext->closeContext(); } + + /** + * Return the prefix of the keys + * + * @return string + */ + protected function getPrefixKeys() + { + $options = $this->pubSubContext->getClient()->getOptions(); + + if (isset($options->prefix)) { + return $options->prefix->getPrefix(); + } + + return ''; + } } diff --git a/lib/Predis/PubSub/PubSubContext.php b/lib/Predis/PubSub/PubSubContext.php index 910459b6..1f4b6b9a 100644 --- a/lib/Predis/PubSub/PubSubContext.php +++ b/lib/Predis/PubSub/PubSubContext.php @@ -41,6 +41,16 @@ class PubSubContext extends AbstractPubSubContext $this->genericSubscribeInit('psubscribe'); } + /** + * Returns the underlying client instance used by the pub/sub iterator. + * + * @return ClientInterface + */ + public function getClient() + { + return $this->client; + } + /** * Checks if the passed client instance satisfies the required conditions * needed to initialize a Publish / Subscribe context. diff --git a/tests/Predis/PubSub/DispatcherLoopTest.php b/tests/Predis/PubSub/DispatcherLoopTest.php index 263469b2..5df84758 100644 --- a/tests/Predis/PubSub/DispatcherLoopTest.php +++ b/tests/Predis/PubSub/DispatcherLoopTest.php @@ -82,4 +82,46 @@ class DispatcherLoopTest extends StandardTestCase $this->assertTrue($consumer->ping()); } + + /** + * @group connected + */ + public function testDispatcherLoopAgainstRedisServerWithPrefix() + { + $parameters = array( + 'host' => REDIS_SERVER_HOST, + 'port' => REDIS_SERVER_PORT, + 'database' => REDIS_SERVER_DBNUM, + // Prevents suite from handing on broken test + 'read_write_timeout' => 2, + ); + + $options = array('profile' => REDIS_SERVER_VERSION); + + $producerNonPfx = new Client($parameters, $options); + $producerNonPfx->connect(); + + $producerPfx = new Client($parameters, $options + array('prefix' => 'foobar')); + $producerPfx->connect(); + + $consumer = new Client($parameters, $options + array('prefix' => 'foobar')); + $dispatcher = new DispatcherLoop($consumer); + + $callback = $this->getMock('stdClass', array('__invoke')); + $callback->expects($this->exactly(1)) + ->method('__invoke') + ->with($this->equalTo('arg:prefixed')) + ->will($this->returnCallback(function ($arg) use ($dispatcher) { + $dispatcher->stop(); + })); + + $dispatcher->attachCallback('callback', $callback); + + $producerNonPfx->publish('callback', 'arg:non-prefixed'); + $producerPfx->publish('callback', 'arg:prefixed'); + + $dispatcher->run(); + + $this->assertTrue($consumer->ping()); + } } diff --git a/tests/Predis/PubSub/PubSubContextTest.php b/tests/Predis/PubSub/PubSubContextTest.php index 5bc36ca2..140e8f25 100644 --- a/tests/Predis/PubSub/PubSubContextTest.php +++ b/tests/Predis/PubSub/PubSubContextTest.php @@ -245,6 +245,19 @@ class PubSubContextTest extends StandardTestCase $this->assertFalse($pubsub->valid()); } + /** + * @group disconnected + */ + public function testGetUnderlyingClientInstance() + { + $connection = $this->getMock('Predis\Connection\SingleConnectionInterface'); + + $client = new Client($connection); + $pubsub = new PubSubContext($client); + + $this->assertSame($client, $pubsub->getClient()); + } + // ******************************************************************** // // ---- INTEGRATION TESTS --------------------------------------------- // // ******************************************************************** //