add: [leave] leaving a channel

Author: gallypette

realtime/channel.py

@@ -40,7 +40,7 @@ def __init__(self, socket: Socket, topic: str, params: Dict[str, Any] = {}) -> N
         self.listeners: List[CallbackListener] = []
         self.joined = False
         self.join_ref = str(uuid.uuid4())
-        self.join_msg_ref = str(uuid.uuid4())
+        self.control_msg_ref = ""
 
     def join(self) -> Channel:
         """
@@ -62,22 +62,23 @@ async def _join(self) -> None:
                         payload={}, ref=None)
         elif self.socket.version == 2:
             #[join_reference, message_reference, topic_name, event_name, payload]
-            join_req = [self.join_ref, self.join_msg_ref, self.topic, ChannelEvents.join, {}]
+            self.control_msg_ref = str(uuid.uuid4())
+            join_req = [self.join_ref, self.control_msg_ref, self.topic, ChannelEvents.join, {}]
 
         try:
             await self.socket.ws_connection.send(json.dumps(join_req))
         except Exception as e:
             print(str(e))  # TODO: better error propagation
             return
 
-    def leave(self) -> Channel:
+    def leave(self) -> None:
         """
         Wrapper for async def _leave() to expose a non-async interface
         Essentially gets the only event loop and attempt leaving a topic
-        :return: Channel
+        :return: None
         """
         loop = asyncio.get_event_loop()  # TODO: replace with get_running_loop
-        loop.run_until_complete(self._join())
+        loop.run_until_complete(self._leave())
         return self
 
     async def _leave(self) -> None:
@@ -86,13 +87,13 @@ async def _leave(self) -> None:
         :return: None
         """
         if self.socket.version == 1:
-            join_req = dict(topic=self.topic, event=ChannelEvents.leave,
+            leave_req = dict(topic=self.topic, event=ChannelEvents.leave,
                         payload={}, ref=None)
         elif self.socket.version == 2:
-            join_req = [self.join_ref, None, self.topic, ChannelEvents.leave, {}]
+            leave_req = [self.join_ref, None, self.topic, ChannelEvents.leave, {}]
 
         try:
-            await self.socket.ws_connection.send(json.dumps(join_req))
+            await self.socket.ws_connection.send(json.dumps(leave_req))
         except Exception as e:
             print(str(e))  # TODO: better error propagation
             return

realtime/connection.py

@@ -81,13 +81,19 @@ async def _listen(self) -> None:
 
                 if msg.event == ChannelEvents.reply:
                     for channel in self.channels.get(msg.topic, []):
-                        if msg.ref == channel.join_msg_ref :
+                        if msg.ref == channel.control_msg_ref :
                             logging.info(f"Successfully joined {msg.topic}")
                             continue
                         else:
                             for cl in channel.listeners:
                                 if cl.ref in ["*", msg.ref]:
                                     cl.callback(msg.payload)
+                
+                if msg.event == ChannelEvents.close:
+                    for channel in self.channels.get(msg.topic, []):
+                        if msg.join_ref == channel.join_ref :
+                            logging.info(f"Successfully left {msg.topic}")
+                            continue
 
                 for channel in self.channels.get(msg.topic, []):
                     for cl in channel.listeners: