From 760ab2dbd61c4edacd709e819057f28e65ee8280 Mon Sep 17 00:00:00 2001
From: Simon Hewitt <si@sjhewitt.co.uk>
Date: Tue, 28 Jul 2015 10:26:06 -0700
Subject: [PATCH] remove called unsubscribe callbacks from list of callbacks

---
 tornadoredis/client.py            | 10 ++++++++--
 tornadoredis/tests/test_pubsub.py | 33 +++++++++++++++++++++++++++++++
 2 files changed, 41 insertions(+), 2 deletions(-)

diff --git a/tornadoredis/client.py b/tornadoredis/client.py
index 0f7fb7e..d469b7e 100644
--- a/tornadoredis/client.py
+++ b/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):
diff --git a/tornadoredis/tests/test_pubsub.py b/tornadoredis/tests/test_pubsub.py
index 527a8ad..348c5e7 100644
--- a/tornadoredis/tests/test_pubsub.py
+++ b/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):