Fix unsubscribe/subscribe of Pubsub + enhance pubsub websocket (#333)

* Fix Pubsub subscribre/unsubscribe

- add missing append to own pool in subscribe
- instead of passing a groupby object to websocket.unsubscribe, we pass a list(vals) (doubled iterating over a groupby object results in an empty List)

* Add missing RECONNECT response

* Update changelog.rst

* Update changelog.rst

* Update pool.py

Author: Dschogo

docs/changelog.rst

@@ -16,6 +16,13 @@ Master
 - ext.commands
     - Bug fixes
         - Make sure double-quotes are properly tokenized for bot commands
+        
+- ext.pubsub
+    - Additions
+        - Websocket automatically handles "RECONNECT" requests by Twitch
+    - Bug fixes      
+        - Unsubscribing from Pubsubevents works again
+
 
 
 - ext.sound

twitchio/ext/pubsub/pool.py

@@ -66,6 +66,7 @@ async def subscribe_topics(self, topics: List[Topic]):
         if node is None:
             node = PubSubWebsocket(self.client, max_topics=self._max_connection_topics)
             await node.connect()
+            self._pool.append(node)
 
         await node.subscribe_topics(topics)
         self._topics.update({t: node for t in topics})
@@ -81,7 +82,7 @@ async def unsubscribe_topics(self, topics: List[Topic]):
 
         """
         for node, vals in itertools.groupby(topics, lambda t: self._topics[t]):
-            await node.unsubscribe_topic(vals)
+            await node.unsubscribe_topic(list(vals))
             if not node.topics:
                 await node.disconnect()
                 self._pool.remove(node)
@@ -91,7 +92,7 @@ def _find_node(self, topics: List[Topic]) -> Optional[PubSubWebsocket]:
             raise ValueError("group is the only supported mode.")
 
         for p in self._pool:
-            if len(p.max_topics) + len(topics) <= p.max_topics:
+            if p.max_topics + len(topics) <= p.max_topics:
                 return p
 
         if len(self._pool) < self._max_size:

twitchio/ext/pubsub/websocket.py

@@ -194,6 +194,9 @@ async def handle_response(self, message: dict):
         if message["error"]:
             logger.error(f"Received errored response for nonce {message['nonce']}: {message['error']}")
             self.client.run_event("pubsub_error", message)
+        elif message["type"] == "RECONNECT":
+            logger.warning("Received RECONNECT response from pubsub edge. Reconnecting")
+            await asyncio.shield(self.reconnect())
         elif message["nonce"]:
             logger.debug(f"Received OK response for nonce {message['nonce']}")
             self.client.run_event("pubsub_nonce", message)