Add types and fix for Python 3.9

README.md

@@ -12,21 +12,21 @@ import phxsocket
 
 Create socket client
 ```python
-socket = phxsocket.Client("wss://target.url/websocket", {"options": "something"})
+socket = phxsocket.Client("wss://target.url/channel/websocket", {"options": "something"})
 ```
 
 Connect and join a channel
 ```python
 if socket.connect(): # blocking, raises exception on failure
   channel = socket.channel("room:roomname", {"more options": "something else"})
-  join_success, resp = channel.join()
+  resp = channel.join() # also blocking, raises exception on failure
 ```
 
 Alternatively
 ```python
 def connect_to_channel(socket):
   channel = socket.channel("room:roomname", {"more options": "something else"})
-  join_success, resp = channel.join()
+  resp = channel.join()
 
 socket.on_open = connect_to_channel
 connection = socket.connect(blocking=False)

phxsocket/channel.py

@@ -1,4 +1,6 @@
 from enum import Enum
+from typing import Union, Callable, Tuple
+from .message import SentMessage
 import traceback
 
 
@@ -23,7 +25,7 @@ def __init__(self, socket, topic, params):
     self.on_close = None
     self.events = {}
 
-  def join(self):
+  def join(self) -> Union[dict, list, str, int, float, bool]:
     join = self.socket.push(self.topic,
                             ChannelEvents.join,
                             self.params,
@@ -35,7 +37,7 @@ def join(self):
 
     return response["response"]
 
-  def leave(self):
+  def leave(self) -> Tuple[bool, Union[dict, list, str, int, float, bool]]:
     leave = self.socket.push(self.topic,
                              ChannelEvents.leave,
                              self.params,
@@ -45,11 +47,17 @@ def leave(self):
     except:
       return False, traceback.format_exc()
 
-  def push(self, event, payload, cb=None, reply=False):
+  def push(self,
+           event: Union[ChannelEvents, str],
+           payload: Union[dict, list, str, int, float, bool],
+           cb: Callable = None,
+           reply: bool = False) -> Union[SentMessage, None]:
     msg = self.socket.push(self.topic, event, payload, cb, reply)
     return msg
 
-  def on(self, event, cb):
+  def on(
+      self, event: Union[ChannelEvents, str],
+      cb: Callable[[Union[dict, list, str, int, float, bool]], None]) -> None:
     self.events[event] = cb
 
   def receive(self, socket, message):

phxsocket/client.py

@@ -1,31 +1,15 @@
 import asyncio, websockets, json, logging, traceback
-from threading import Event, Thread, Lock
+from threading import Thread, Lock
 from urllib.parse import urlencode
 from .channel import Channel, ChannelEvents
-from .message import Message
+from .message import Message, SentMessage
+from typing import Union, Callable
 
 
 class SocketClosedError(Exception):
   pass
 
 
-class SentMessage:
-  def __init__(self, cb=None):
-    self.cb = cb
-    self.event = Event()
-    self.message = None
-
-  def respond(self, message):
-    self.message = message
-    if self.cb:
-      self.cb(message)
-    self.event.set()
-
-  def wait_for_response(self):
-    self.event.wait()
-    return self.message
-
-
 class ClientConnection(SentMessage):
   def __init__(self, client):
     super().__init__()
@@ -55,7 +39,7 @@ def is_set(self):
 
 
 class Client:
-  def __init__(self, url, params):
+  def __init__(self, url: str, params: dict = {}):
     self._url = url
     self.set_params(params)
     self._loop = None
@@ -76,7 +60,7 @@ def __init__(self, url, params):
 
     self._send_queue = None
 
-  def set_params(self, params, url=None):
+  def set_params(self, params: dict = {}, url: str = None) -> None:
     qs_params = {"vsn": "1.0.0", **params}
     if url:
       self._url = url
@@ -99,6 +83,8 @@ async def _broadcast(self, websocket, send_queue):
         if message:
           await websocket.send(message)
         send_queue.task_done()
+    except asyncio.exceptions.CancelledError:
+      logging.info("phxsocket: broadcast queue finished")
     except:
       logging.error("phxsocket: FATAL ERROR: " + traceback.format_exc())
 
@@ -130,7 +116,7 @@ def run(self, connect_evt):
       else:
         logging.error("phxsocket: " + traceback.format_exc())
     finally:
-      for task in asyncio.Task.all_tasks(loop):
+      for task in asyncio.all_tasks(loop):
         task.cancel()
 
       # notify self._broadcast
@@ -141,14 +127,14 @@ def run(self, connect_evt):
       if connect_evt.is_set() and self.on_close:
         self.on_close(self)
 
-  def close(self):
+  def close(self) -> None:
     if not self._loop:
       raise SocketClosedError
 
     self._loop.call_soon_threadsafe(self._shutdown_evt.set)
     self.thread.join()
 
-  def connect(self, blocking=True):
+  def connect(self, blocking: bool = True) -> Union[ClientConnection, None]:
     if self._loop:
       logging.error("phxsocket: Trying to start another thread")
       return False
@@ -180,7 +166,12 @@ def _on_message(self, _message):
     if self.on_message:
       Thread(target=self.on_message, args=[message], daemon=True).start()
 
-  def push(self, topic, event, payload, cb=None, reply=False):
+  def push(self,
+           topic: str,
+           event: Union[ChannelEvents, str],
+           payload: Union[dict, list, str, int, float, bool],
+           cb: Callable = None,
+           reply: bool = False) -> Union[SentMessage, None]:
     if not self._loop:
       raise SocketClosedError
 
@@ -208,7 +199,7 @@ def push(self, topic, event, payload, cb=None, reply=False):
     if reply or cb:
       return sent_message
 
-  def channel(self, topic, params={}):
+  def channel(self, topic: str, params: dict = {}) -> Channel:
     if topic not in self.channels:
       channel = Channel(self, topic, params)
       self.channels[topic] = channel

phxsocket/message.py

@@ -1,4 +1,5 @@
 import json, typing
+from threading import Event
 
 
 class Message(typing.NamedTuple):
@@ -11,3 +12,20 @@ class Message(typing.NamedTuple):
   def from_json(cls, msg):
     msg = json.loads(msg)
     return cls(msg["event"], msg["topic"], msg["payload"], msg["ref"])
+
+
+class SentMessage:
+  def __init__(self, cb=None):
+    self.cb = cb
+    self.event = Event()
+    self.message = None
+
+  def respond(self, message):
+    self.message = message
+    if self.cb:
+      self.cb(message)
+    self.event.set()
+
+  def wait_for_response(self):
+    self.event.wait()
+    return self.message

setup.py

@@ -8,7 +8,7 @@
 
 setup(
   name="phxsocket",
-  version="0.1.0",
+  version="0.1.2",
   author="wwwwwwww",
   author_email="[email protected]",
   description="Websocket client for Phoenix Elixir",
@@ -22,5 +22,5 @@
     "License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
     "Operating System :: OS Independent",
   ],
-  python_requires=">=3.6",
+  python_requires=">=3.7",
 )