add optional "timeout" parameter to pubsub.can_read

redis/client.py

@@ -2142,10 +2142,10 @@ def _execute(self, connection, command, *args):
             # previously listening to
             return command(*args)
 
-    def parse_response(self, block=True):
+    def parse_response(self, block=True, timeout=0):
         "Parse the response from a publish/subscribe command"
         connection = self.connection
-        if not block and not connection.can_read():
+        if not block and not connection.can_read(timeout=timeout):
             return None
         return self._execute(connection, connection.read_response)
 
@@ -2216,9 +2216,9 @@ def listen(self):
             if response is not None:
                 yield response
 
-    def get_message(self, ignore_subscribe_messages=False):
+    def get_message(self, ignore_subscribe_messages=False, timeout=0):
         "Get the next message if one is available, otherwise None"
-        response = self.parse_response(block=False)
+        response = self.parse_response(block=False, timeout=timeout)
         if response:
             return self.handle_message(response, ignore_subscribe_messages)
         return None

redis/connection.py

@@ -555,13 +555,13 @@ def send_command(self, *args):
         "Pack and send a command to the Redis server"
         self.send_packed_command(self.pack_command(*args))
 
-    def can_read(self):
+    def can_read(self, timeout=0):
         "Poll the socket to see if there's data that can be read."
         sock = self._sock
         if not sock:
             self.connect()
             sock = self._sock
-        return bool(select([sock], [], [], 0)[0]) or self._parser.can_read()
+        return self._parser.can_read() or bool(select([sock], [], [], timeout)[0])
 
     def read_response(self):
         "Read the response from a previously sent command"