Add unsubscribe support

Author: nneskildsf

fastapi_websocket_pubsub/pub_sub_client.py

@@ -280,6 +280,47 @@ def subscribe(self, topic: Topic, callback: Coroutine):
             future.set_result(None)
             return future
 
+    def unsubscribe(self, topic: Topic):
+        """
+        Unsubscribe for events
+
+        Args:
+            topic (Topic): the identifier of the event topic to be unsubscribed.
+                           Note: You can use ALL_TOPICS (event_notifier.ALL_TOPICS) to unsubscribe all topics
+
+        Returns:
+            Coroutine: awaitable task to subscribe to topic if connected.
+        """
+        # Create none-future which can be safely awaited
+        # but which also will not give warnings
+        # if it isn't awaited. This is returned
+        # on code paths which do not make RPC calls.
+        none_future = asyncio.Future()
+        none_future.set_result(None)
+
+        # Handle unsubscribing ALL_TOPICS
+        if topic is ALL_TOPICS and not self._topics:
+            logger.warning(f"Cannot unsubscribe 'ALL_TOPICS'. No topics are subscribed.")
+            return none_future
+        elif topic is not ALL_TOPICS and topic not in self._topics:
+            logger.warning(f"Cannot unsubscribe topic '{topic}' which is not subscribed.")
+            return none_future
+        elif topic is ALL_TOPICS and self._topics:
+            logger.debug(f"Unsubscribing all topics: {self._topics}")
+            # remove all topics and callbacks
+            self._topics.clear()
+            self._callbacks.clear()
+        elif topic is not ALL_TOPICS and topic in self._topics:
+            logger.debug(f"Unsubscribing topic '{topic}'")
+            self._topics.remove(topic)
+            self._callbacks.pop(topic, None)
+        
+        if self.is_ready():
+            topics = list(self._topics) if topic is ALL_TOPICS else [topic]
+            return self._rpc_channel.other.unsubscribe(topics=topics)
+        else:
+            return none_future
+
     async def publish(
         self, topics: TopicList, data=None, sync=True, notifier_id=None
     ) -> bool: