Merge branch 'pr/1033' into monitor

3 files changed, 71 insertions, 0 deletions

diff --git a/README.rst b/README.rst
index 8726d39..01551fa 100644
--- a/README.rst
+++ b/README.rst
@@ -675,6 +675,18 @@ supported:
     >>> r.pubsub_numpat()
     1204
 
+Monitor
+^^^^^^^
+redis-py includes a `Monitor` object that that streams back every command
+processed by the Redis server. Use `listen` on the `Monitor` object to block
+until message available.
+
+.. code-block:: pycon
+
+    >>> r = redis.StrictRedis(...)
+    >>> with sr.monitor() as m:
+    >>>   for command in m.listen():
+    >>>     print(command)
 
 Lua Scripting
 ^^^^^^^^^^^^^
diff --git a/redis/client.py b/redis/client.py
index de9cf88..07df19a 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -6,6 +6,7 @@ import warnings
 import time
 import threading
 import time as mod_time
+import re
 import hashlib
 from redis._compat import (basestring, imap, iteritems, iterkeys,
                            itervalues, izip, long, nativestr, safe_unicode)
@@ -764,6 +765,9 @@ class Redis(object):
         """
         return PubSub(self.connection_pool, **kwargs)
 
+    def monitor(self):
+        return Monitor(self.connection_pool)
+
     # COMMAND EXECUTION AND PROTOCOL PARSING
     def execute_command(self, *args, **options):
         "Execute a command and return a parsed response"
@@ -2922,6 +2926,51 @@ class Redis(object):
 StrictRedis = Redis
 
 
+class Monitor(object):
+    """
+    Monitor is useful for handling the MONITOR command to the redis server.
+    next_command() method returns one command from monitor
+    listen() method yields commands from monitor.
+    """
+    def __init__(self, connection_pool):
+        self.connection_pool = connection_pool
+        self.connection = self.connection_pool.get_connection('MONITOR')
+
+    def __enter__(self):
+        self.connection.send_command('MONITOR')
+        # check that monitor returns 'OK', but don't return it to user
+        response = self.connection.read_response()
+        if not bool_ok(response):
+            raise RedisError('MONITOR failed: %s' % response)
+        return self
+
+    def __exit__(self, *args):
+        self.connection.disconnect()
+        self.connection_pool.release(self.connection)
+
+    def next_command(self):
+        "Parse the response from a monitor command"
+        response = self.connection.read_response()
+        if isinstance(response, bytes):
+            response = self.connection.encoder.decode(response, force=True)
+        command_time, command_data = response.split(' ', 1)
+        m = re.match(r'\[(\d+) (.+):(\d+)\] (.*)', command_data)
+        db_id, client_address, client_port, command = m.groups()
+        command = re.match(r'"(\w*)"+', command).groups()
+        return {
+            'time': float(command_time),
+            'db': db_id,
+            'client_address': client_address,
+            'client_port': client_port,
+            'command': command
+        }
+
+    def listen(self):
+        "Listen for commands coming to the server."
+        while 1:
+            yield self.next_command()
+
+
 class PubSub(object):
     """
     PubSub provides publish, subscribe and listen support to Redis channels.
diff --git a/tests/test_monitor.py b/tests/test_monitor.py
new file mode 100644
index 0000000..da0c992
--- /dev/null
+++ b/tests/test_monitor.py
@@ -0,0 +1,10 @@
+from __future__ import unicode_literals
+
+
+class TestPipeline(object):
+    def test_monitor(self, r):
+        with r.monitor() as m:
+            r.ping()
+            response = m.next_command()
+            assert set(response.keys()) == {'time', 'db', 'client_address',
+                                            'client_port', 'command'}