WIP making tests pass with the new async servers

Author: tshirtman

examples/asyncio_example.py

@@ -6,6 +6,7 @@
 async def osc_app(address, port):
     osc = OSCAsyncioServer(encoding='utf8')
     osc.listen(address=address, port=port, default=True)
+    sock2 = osc.listen(address=address, port=port + 1)
 
     @osc.address("/example")
     async def example(*values):
@@ -19,11 +20,12 @@ async def test(*values):
         await asyncio.sleep(4)
         print("done sleeping")
 
-    @osc.address("/stop")
+    @osc.address("/stop", sock=sock2)
     async def stop(*values):
         print(f"time to leave!")
         osc.terminate_server()
 
+    print(sock2.getsockname())
     asyncio.get_event_loop().create_task(osc.process())
     await osc.join_server()
 

oscpy/server/__init__.py

@@ -522,7 +522,7 @@ def _execute_callbacks(self, callbacks_list, address, values):
                     cb(*values)
             except Exception:
                 if self.intercept_errors:
-                    logger.error("Unhandled exception caught in oscpy server", exc_info=True)
+                    logger.exception("Unhandled exception caught in oscpy server")
                 else:
                     raise
 
@@ -577,7 +577,6 @@ def join_server(self, timeout=None):
         Returns True if and only if the inner thread exited before timeout."""
         return self._termination_event.wait(timeout=timeout)
 
-
 # backward compatibility
 
 from oscpy.server.thread_server import OSCThreadServer

oscpy/server/asyncio_server.py

@@ -1,10 +1,14 @@
 import asyncio
 import socket
 from functools import partial
+from logging import getLogger
 
 from oscpy.server import OSCBaseServer
 
 
+logger = getLogger(__name__)
+
+
 class OSCAsyncioServer(OSCBaseServer):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -13,17 +17,20 @@ def __init__(self, *args, **kwargs):
 
     def listen(self, address='localhost', port=0, default=False, family='inet', **kwargs):
         loop = asyncio.get_event_loop()
-        addr = (address, port)
+        if family == 'unix':
+            addr = address
+        else:
+            addr = (address, port)
         sock = self.get_socket(
             family=socket.AF_UNIX if family == 'unix' else socket.AF_INET,
             addr=addr,
         )
-        self.listeners[addr] = awaitable = loop.create_datagram_endpoint(
+        self.listeners[(address, port or sock.getsockname()[1])] = loop.create_datagram_endpoint(
             partial(OSCProtocol, self.handle_message, sock),
             sock=sock,
         )
         self.add_socket(sock, default)
-        return awaitable
+        return sock
 
     async def process(self):
         return await asyncio.gather(
@@ -36,7 +43,6 @@ async def handle_message(self, data, sender, sender_socket):
             await self._execute_callbacks(callbacks, address, values)
 
     async def _execute_callbacks(self, callbacks_list, address, values):
-        print(locals())
         for cb, get_address in callbacks_list:
             try:
                 if get_address:
@@ -45,10 +51,29 @@ async def _execute_callbacks(self, callbacks_list, address, values):
                     await cb(*values)
             except Exception:
                 if self.intercept_errors:
-                    logger.error("Unhandled exception caught in oscpy server", exc_info=True)
+                    logger.exception("Unhandled exception caught in oscpy server")
                 else:
                     raise
 
+    def stop(self, sock=None):
+        """Close and remove a socket from the server's sockets.
+
+        If `sock` is None, uses the default socket for the server.
+
+        """
+        if not sock and self.default_socket:
+            sock = self.default_socket
+
+        if sock in self.sockets:
+            sock.close()
+            self.sockets.remove(sock)
+        else:
+            raise RuntimeError('{} is not one of my sockets!'.format(sock))
+
+    def stop_all(self):
+        for sock in self.sockets[:]:
+            self.stop(sock)
+
     async def join_server(self, timeout=None):
         """Wait for the server to exit (`terminate_server()` must have been called before).
 
@@ -60,13 +85,16 @@ class OSCProtocol(asyncio.DatagramProtocol):
     def __init__(self, message_handler, sock, **kwargs):
         super().__init__(**kwargs)
         self.message_handler = message_handler
-        self._socket = sock
+        self.socket = sock
         self.loop = asyncio.get_event_loop()
 
     def connection_made(self, transport):
         self.transport = transport
 
     def datagram_received(self, data, addr):
         self.loop.call_soon(
-            lambda: asyncio.ensure_future(self.message_handler(data, addr, self._socket))
+            lambda: asyncio.ensure_future(self.message_handler(data, addr, self.socket))
         )
+
+    def getsockname(self):
+        return self.socket.getsockname()

oscpy/server/curio_server.py

@@ -61,6 +61,12 @@ async def process(self):
     async def stop_all(self):
         await self.tasks_group.cancel_remaining()
 
-    async def stop(self, sock):
-        g = self.task_groups.pop(sock)
-        await g.cancel_remaining()
+    async def stop(self, sock=None):
+        if not sock and self.default_socket:
+            sock = self.default_socket
+
+        if sock in self.sockets:
+            g = self.task_groups.pop(sock)
+            await g.cancel_remaining()
+        else:
+            raise RuntimeError('{} is not one of my sockets!'.format(sock))

oscpy/server/thread_server.py

@@ -83,3 +83,8 @@ def _listen(self):
                     continue
 
                 self.handle_message(data, sender, sender_socket)
+
+    def join_server(self, timeout=None):
+        result = super(OSCThreadServer, self).join_server(timeout=timeout)
+        self._thread.join(timeout=timeout)
+        return result