Merge pull request clue#23 from clue/monitor

Support MONITOR command, forward "monitor" events

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

@@ -11,6 +11,8 @@
  *
  * @event data(ModelInterface $messageModel, Client $thisClient)
  * @event close()
+ *
+ * @event monitor(ModelInterface $statusModel)
  */
 interface Client extends EventEmitterInterface
 {

src/StreamingClient.php

@@ -14,6 +14,7 @@
 use React\Promise\Deferred;
 use Clue\Redis\Protocol\Model\ErrorReply;
 use Clue\Redis\Protocol\Model\ModelInterface;
+use Clue\Redis\Protocol\Model\StatusReply;
 
 class StreamingClient extends EventEmitter implements Client
 {
@@ -24,6 +25,8 @@ class StreamingClient extends EventEmitter implements Client
     private $ending = false;
     private $closed = false;
 
+    private $monitoring = false;
+
     public function __construct(Stream $stream, ParserInterface $parser = null, SerializerInterface $serializer = null)
     {
         if ($parser === null || $serializer === null) {
@@ -78,13 +81,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->requests) {
             throw new UnderflowException('Unexpected reply received, no matching request found');
         }
@@ -137,4 +152,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

@@ -103,6 +103,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

@@ -5,6 +5,7 @@
 use Clue\Redis\Protocol\Model\IntegerReply;
 use Clue\Redis\Protocol\Model\BulkReply;
 use Clue\Redis\Protocol\Model\ErrorReply;
+use Clue\Redis\Protocol\Model\StatusReply;
 
 class ClientTest extends TestCase
 {
@@ -91,6 +92,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();