Merge branch 'github/pr/114' into v0.8

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 '';
+    }
 }

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.

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());
+    }
 }

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 --------------------------------------------- //
     // ******************************************************************** //