remove called unsubscribe callbacks from list of callbacks

tornadoredis/client.py

@@ -1064,9 +1064,15 @@ def on_subscribed(self, result):
     def on_unsubscribed(self, channels, *args, **kwargs):
         channels = set(channels)
         self.subscribed -= channels
-        for cb_channels, cb in self.unsubscribe_callbacks:
+        callbacks = self.unsubscribe_callbacks
+        self.unsubscribe_callbacks = []
+        for cb_channels, cb in callbacks:
             cb_channels.difference_update(channels)
-            if not cb_channels:
+            # the callback is only called when the client has been unsubscribed
+            # all of the channels
+            if cb_channels:
+                self.unsubscribe_callbacks.append((cb_channels, cb))
+            else:
                 self._io_loop.add_callback(cb)
 
     def unsubscribe(self, channels, callback=None):

tornadoredis/tests/test_pubsub.py

@@ -88,6 +88,39 @@ def on_message(*args, **kwargs):
         self.assertEqual(self._message_count, 4)
         self.stop()
 
+    @async_test
+    @gen.engine
+    def test_unsubscribe_callbacks(self):
+        def on_message(*args, **kwargs):
+            self._message_count += 1
+
+        yield gen.Task(self.client.subscribe, 'foo')
+        self.client.listen(on_message, (yield gen.Callback('listen')))
+        self.assertTrue(self.client.subscribed)
+
+        cb_check = {}
+
+        @gen.engine
+        def on_unsubscribe1(*args, **kwargs):
+            self.assertFalse(cb_check.get('foo'))
+            self.assertFalse(cb_check.get('bar'))
+            self.assertFalse(self.client.subscribed)
+            cb_check['foo'] = True
+            yield gen.Task(self.client.subscribe, 'bar')
+            self.client.listen(on_message, (yield gen.Callback('listen')))
+            self.assertTrue(self.client.subscribed)
+            self.client.unsubscribe('bar', on_unsubscribe2)
+            yield gen.Wait('listen')
+
+        def on_unsubscribe2(*args, **kwargs):
+            self.assertTrue(cb_check.get('foo'))
+            self.assertFalse(cb_check.get('bar'))
+            cb_check['bar'] = True
+            self.stop()
+
+        self.client.unsubscribe('foo', on_unsubscribe1)
+        yield gen.Wait('listen')
+
     @async_test
     @gen.engine
     def test_pub_sub_multiple(self):