add: [send] listener on a msg ref

Author: gallypette

poetry.lock

@@ -9,6 +9,7 @@ python = "^3.8"
 websockets = "^11.0"
 python-dateutil = "^2.8.1"
 typing-extensions = "^4.2.0"
+uuid = "^1.30"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.2.0"

pyproject.toml

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import uuid
 from typing import Any, List, Dict, TYPE_CHECKING, NamedTuple
 from realtime.message import *
 
@@ -14,13 +15,15 @@
 class CallbackListener(NamedTuple):
     """A tuple with `event` and `callback` """
     event: str
+    ref: str
     callback: Callback
 
 
 class Channel:
     """
     `Channel` is an abstraction for a topic listener for an existing socket connection.
     Each Channel has its own topic and a list of event-callbacks that responds to messages.
+    A client can also send messages to a channel and register callback when expecting replies.
     Should only be instantiated through `connection.Socket().set_channel(topic)`
     Topic-Channel has a 1-many relationship.
     """
@@ -36,6 +39,8 @@ def __init__(self, socket: Socket, topic: str, params: Dict[str, Any] = {}) -> N
         self.topic = topic
         self.listeners: List[CallbackListener] = []
         self.joined = False
+        self.join_ref = str(uuid.uuid4())
+        self.join_msg_ref = str(uuid.uuid4())
 
     def join(self) -> Channel:
         """
@@ -56,7 +61,8 @@ async def _join(self) -> None:
             join_req = dict(topic=self.topic, event=ChannelEvents.join,
                         payload={}, ref=None)
         elif self.socket.version == 2:
-            join_req = [None, None, self.topic, ChannelEvents.join, {}]
+            #[join_reference, message_reference, topic_name, event_name, payload]
+            join_req = [self.join_ref, self.join_msg_ref, self.topic, ChannelEvents.join, {}]
 
         try:
             await self.socket.ws_connection.send(json.dumps(join_req))
@@ -83,60 +89,64 @@ async def _leave(self) -> None:
             join_req = dict(topic=self.topic, event=ChannelEvents.leave,
                         payload={}, ref=None)
         elif self.socket.version == 2:
-            join_req = [None, None, self.topic, ChannelEvents.leave, {}]
+            join_req = [self.join_ref, None, self.topic, ChannelEvents.leave, {}]
 
         try:
             await self.socket.ws_connection.send(json.dumps(join_req))
         except Exception as e:
             print(str(e))  # TODO: better error propagation
             return
 
-    def on(self, event: str, callback: Callback) -> Channel:
+    def on(self, event: str, ref: str, callback: Callback) -> Channel:
         """
         :param event: A specific event will have a specific callback
+        :param ref: A specific reference that will have a specific callback
         :param callback: Callback that takes msg payload as its first argument
         :return: Channel
         """
-        cl = CallbackListener(event=event, callback=callback)
+        cl = CallbackListener(event=event, ref=ref, callback=callback)
         self.listeners.append(cl)
         return self
 
-    def off(self, event: str) -> None:
+    def off(self, event: str, ref: str) -> None:
         """
         :param event: Stop responding to a certain event
+        :param event: Stop responding to a certain reference
         :return: None
         """
         self.listeners = [
-            callback for callback in self.listeners if callback.event != event]
+            callback for callback in self.listeners if (callback.event != event and callback.ref != ref)]
 
-    def send(self, event_name: str, payload:str) -> None:
+    def send(self, event_name: str, payload:str, ref: uuid = str(uuid.uuid4())) -> None:
         """
         Wrapper for async def _send() to expose a non-async interface
         Essentially gets the only event loop and attempt sending a payload
         to a topic
         :param event_name: The event_name: it must match the first argument of a handle_in function on the server channel module.
         :param payload: The payload to be sent to the phoenix server
+        :param ref: The message reference that the server will use for replying - if none is set, generates the string repr of a uuidv4
         :return: None
         """
         loop = asyncio.get_event_loop()  # TODO: replace with get_running_loop
-        loop.run_until_complete(self._send(event_name, payload))
+        loop.run_until_complete(self._send(event_name, payload, ref))
         return self
 
-    async def _send(self, event_name: str, payload:str) -> None:
+    async def _send(self, event_name: str, payload: str, ref: str) -> None:
         """
         Coroutine that attempts to join Phoenix Realtime server via a certain topic
         :param event_name: The event_name: it must match the first argument of a handle_in function on the server channel module.
         :param payload: The payload to be sent to the phoenix server
+        :param ref: The message reference that the server will use for replying
         :return: None
         """
         if self.socket.version == 1:
             msg = dict(topic=self.topic, event=event_name,
                         payload=payload, ref=None)
         elif self.socket.version == 2:
-            msg = [3, 3, self.topic, event_name, payload]
+            msg = [None, ref, self.topic, event_name, payload]
 
         try:
             await self.socket.ws_connection.send(json.dumps(msg))
         except Exception as e:
             print(str(e))  # TODO: better error propagation
-            return
+            return

realtime/channel.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import logging
+import pdb
 from collections import defaultdict
 from functools import wraps
 from typing import Any, Callable, List, Dict, TypeVar, DefaultDict
@@ -76,15 +77,23 @@ async def _listen(self) -> None:
                     msg = Message(**json.loads(msg))
                 elif self.version == 2:
                     msg_array = json.loads(msg)
-                    msg = Message(chanid=msg_array[0], ref= msg_array[1], topic=msg_array[2], event= msg_array[3], payload= msg_array[4])
+                    msg = Message(join_ref=msg_array[0], ref= msg_array[1], topic=msg_array[2], event= msg_array[3], payload= msg_array[4])
 
                 if msg.event == ChannelEvents.reply:
-                    continue
+                    for channel in self.channels.get(msg.topic, []):
+                        if msg.ref == channel.join_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)
 
                 for channel in self.channels.get(msg.topic, []):
                     for cl in channel.listeners:
                         if cl.event in ["*", msg.event]:
                             cl.callback(msg.payload)
+
             except websockets.exceptions.ConnectionClosed:
                 if self.auto_reconnect:
                     logging.info("Connection with server closed, trying to reconnect...")
@@ -155,10 +164,10 @@ def set_channel(self, topic: str) -> Channel:
 
     def summary(self) -> None:
         """
-        Prints a list of topics and event the socket is listening to
+        Prints a list of topics and event, and reference that the socket is listening to
         :return: None
         """
         for topic, chans in self.channels.items():
             for chan in chans:
                 print(
-                    f"Topic: {topic} | Events: {[e for e, _ in chan.listeners]}]")
+                    f"Topic: {topic} | Events: {[e for e, _, _ in chan.listeners]} | References: {[r for _, r, _ in chan.listeners]}]")

realtime/connection.py

@@ -11,7 +11,7 @@ class Message:
     event: str
     payload: Dict[str, Any]
     ref: Any
-    chanid: Any
+    join_ref: Any
     topic: str
 
     def __hash__(self):