Merge branch 'master' into pubsub

Author: clue

examples/monitor.php

@@ -0,0 +1,24 @@
+<?php
+
+use Clue\React\Redis\Client;
+use Clue\React\Redis\Factory;
+use Clue\Redis\Protocol\Model\StatusReply;
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$loop = React\EventLoop\Factory::create();
+$factory = new Factory($loop);
+
+$factory->createClient()->then(function (Client $client) {
+    $client->monitor()->then(function ($result) {
+        echo 'Now monitoring all commands' . PHP_EOL;
+    });
+
+    $client->on('monitor', function (StatusReply $message) {
+        echo 'Monitored: ' . $message->getValueNative() . PHP_EOL;
+    });
+
+    $client->echo('initial echo');
+});
+
+$loop->run();

src/Client.php

@@ -19,6 +19,8 @@
  * @event pmessage($pattern, $channel, $message)
  * @event psubscribe($channel, $numberOfChannels)
  * @event punsubscribe($channel, $numberOfChannels)
+ *
+ * @event monitor(ModelInterface $statusModel)
  */
 interface Client extends EventEmitterInterface
 {

src/StreamingClient.php

@@ -15,6 +15,7 @@
 use Clue\Redis\Protocol\Model\ErrorReply;
 use Clue\Redis\Protocol\Model\ModelInterface;
 use Clue\Redis\Protocol\Model\MultiBulkReply;
+use Clue\Redis\Protocol\Model\StatusReply;
 
 class StreamingClient extends EventEmitter implements Client
 {
@@ -27,6 +28,7 @@ class StreamingClient extends EventEmitter implements Client
 
     private $subscribed = 0;
     private $psubscribed = 0;
+    private $monitoring = false;
 
     public function __construct(Stream $stream, ParserInterface $parser = null, SerializerInterface $serializer = null)
     {
@@ -82,13 +84,25 @@ public function __call($name, $args)
             $this->requests []= $request;
         }
 
+        if (strtolower($name) === 'monitor') {
+            $monitoring =& $this->monitoring;
+            $request->then(function () use (&$monitoring) {
+                $monitoring = true;
+            });
+        }
+
         return $request->promise();
     }
 
     public function handleMessage(ModelInterface $message)
     {
         $this->emit('data', array($message, $this));
 
+        if ($this->monitoring && $this->isMonitorMessage($message)) {
+            $this->emit('monitor', array($message));
+            return;
+        }
+
         if (/*($this->subscribed !== 0 || $this->psubscribed !== 0) &&*/ $message instanceof MultiBulkReply) {
             $array = $message->getValueNative();
             $first = array_shift($array);
@@ -156,4 +170,10 @@ public function close()
             $request->reject(new RuntimeException('Connection closing'));
         }
     }
+
+    private function isMonitorMessage(ModelInterface $message)
+    {
+        // Check status '1409172115.207170 [0 127.0.0.1:58567] "ping"' contains otherwise uncommon '] "'
+        return ($message instanceof StatusReply && strpos($message->getValueNative(), '] "') !== false);
+    }
 }

tests/FunctionalTest.php

@@ -117,6 +117,21 @@ public function testMultiExecQueuedExecHasValues(StreamingClient $client)
         return $client;
     }
 
+    /**
+     *
+     * @param StreamingClient $client
+     * @depends testPipeline
+     */
+    public function testMonitorPing(StreamingClient $client)
+    {
+        $client->on('monitor', $this->expectCallableOnce());
+
+        $client->monitor()->then($this->expectCallableOnce('OK'));
+        $client->ping()->then($this->expectCallableOnce('PONG'));
+
+        $this->waitFor($client);
+    }
+
     public function testPubSub()
     {
         $consumer = $this->createClient();

tests/StreamingClientTest.php

@@ -7,6 +7,7 @@
 use Clue\Redis\Protocol\Model\ErrorReply;
 use Clue\Redis\Protocol\Model\MultiBulkReply;
 use Clue\React\Redis\Client;
+use Clue\Redis\Protocol\Model\StatusReply;
 
 class ClientTest extends TestCase
 {
@@ -93,6 +94,52 @@ public function testPingPong()
         $promise->then($this->expectCallableOnce('PONG'));
     }
 
+    /**
+     * @expectedException UnderflowException
+     */
+    public function testInvalidMonitor()
+    {
+        $this->client->handleMessage(new StatusReply('+1409171800.312243 [0 127.0.0.1:58542] "ping"'));
+    }
+
+    public function testMonitor()
+    {
+        $this->serializer->expects($this->once())->method('getRequestMessage')->with($this->equalTo('monitor'));
+
+        $promise = $this->client->monitor();
+
+        $this->client->handleMessage(new StatusReply('OK'));
+
+        $this->expectPromiseResolve($promise);
+        $promise->then($this->expectCallableOnce('OK'));
+
+        return $this->client;
+    }
+
+    /**
+     * @depends testMonitor
+     * @param StreamingClient $client
+     */
+    public function testMonitorEvent(StreamingClient $client)
+    {
+        $client->on('monitor', $this->expectCallableOnce());
+
+        $client->handleMessage(new StatusReply('1409171800.312243 [0 127.0.0.1:58542] "ping"'));
+    }
+
+    /**
+     * @depends testMonitor
+     * @param StreamingClient $client
+     */
+    public function testMonitorPing(StreamingClient $client)
+    {
+        $client->on('monitor', $this->expectCallableOnce());
+
+        $client->ping();
+        $client->handleMessage(new StatusReply('1409171800.312243 [0 127.0.0.1:58542] "ping"'));
+        $client->handleMessage(new StatusReply('PONG'));
+    }
+
     public function testErrorReply()
     {
         $promise = $this->client->invalid();