Handle unix sockets in UDP

Fixes #269.

Co-authored-by: Dmitry Orlov <me@mosquito.su>

tests/test_udp.py

@@ -161,18 +161,70 @@ def test_create_datagram_endpoint_sock(self):
     @unittest.skipIf(sys.version_info < (3, 5, 1),
                      "asyncio in 3.5.0 doesn't have the 'sock' argument")
     def test_create_datagram_endpoint_sock_unix_domain(self):
+
+        class Proto(asyncio.DatagramProtocol):
+            done = None
+
+            def __init__(self, loop):
+                self.state = 'INITIAL'
+                self.addrs = set()
+                self.done = asyncio.Future(loop=loop)
+                self.data = b''
+
+            def connection_made(self, transport):
+                self.transport = transport
+                assert self.state == 'INITIAL', self.state
+                self.state = 'INITIALIZED'
+
+            def datagram_received(self, data, addr):
+                assert self.state == 'INITIALIZED', self.state
+                self.addrs.add(addr)
+                self.data += data
+                if self.data == b'STOP' and not self.done.done():
+                    self.done.set_result(True)
+
+            def error_received(self, exc):
+                assert self.state == 'INITIALIZED', self.state
+                if not self.done.done():
+                    self.done.set_exception(exc or RuntimeError())
+
+            def connection_lost(self, exc):
+                assert self.state == 'INITIALIZED', self.state
+                self.state = 'CLOSED'
+                if self.done and not self.done.done():
+                    self.done.set_result(None)
+
         tmp_file = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
         sock = socket.socket(socket.AF_UNIX, type=socket.SOCK_DGRAM)
         sock.bind(tmp_file)
 
         with sock:
+            pr = Proto(loop=self.loop)
             f = self.loop.create_datagram_endpoint(
-                lambda: MyDatagramProto(loop=self.loop), sock=sock)
-            tr, pr = self.loop.run_until_complete(f)
-            self.assertIsInstance(pr, MyDatagramProto)
-            tr.sendto(b'HELLO', tmp_file)
-            tr.close()
-            self.loop.run_until_complete(pr.done)
+                lambda: pr, sock=sock)
+            tr, pr_prime = self.loop.run_until_complete(f)
+            self.assertIs(pr, pr_prime)
+
+            tmp_file2 = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
+            sock2 = socket.socket(socket.AF_UNIX, type=socket.SOCK_DGRAM)
+            sock2.bind(tmp_file2)
+
+            with sock2:
+                f2 = self.loop.create_datagram_endpoint(
+                    asyncio.DatagramProtocol, sock=sock2)
+                tr2, pr2 = self.loop.run_until_complete(f2)
+
+                tr2.sendto(b'STOP', tmp_file)
+
+                self.loop.run_until_complete(pr.done)
+
+                tr.close()
+                tr2.close()
+
+                # Let transports close
+                self.loop.run_until_complete(asyncio.sleep(0.2))
+
+                self.assertIn(tmp_file2, pr.addrs)
 
 
 class Test_UV_UDP(_TestUDP, tb.UVTestCase):

uvloop/dns.pyx

@@ -33,6 +33,7 @@ cdef __convert_sockaddr_to_pyaddr(const system.sockaddr* addr):
         int err
         system.sockaddr_in *addr4
         system.sockaddr_in6 *addr6
+        system.sockaddr_un *addr_un
 
     if addr.sa_family == uv.AF_INET:
         addr4 = <system.sockaddr_in*>addr
@@ -60,6 +61,10 @@ cdef __convert_sockaddr_to_pyaddr(const system.sockaddr* addr):
             addr6.sin6_scope_id
         )
 
+    elif addr.sa_family == uv.AF_UNIX:
+        addr_un = <system.sockaddr_un*>addr
+        return system.MakeUnixSockPyAddr(addr_un)
+
     raise RuntimeError("cannot convert sockaddr into Python object")
 
 

uvloop/includes/compat.h

@@ -1,4 +1,7 @@
 #include <errno.h>
+#include <stddef.h>
+#include <sys/socket.h>
+#include <sys/un.h>
 #include "Python.h"
 #include "uv.h"
 
@@ -21,10 +24,36 @@
 #  define PLATFORM_IS_LINUX 0
 #  define EPOLL_CTL_DEL 2
 struct epoll_event {};
-int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) {};
+int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) {
+    return 0;
+};
 #endif
 
 
+PyObject *
+MakeUnixSockPyAddr(struct sockaddr_un *addr)
+{
+    if (addr->sun_family != AF_UNIX) {
+        PyErr_SetString(
+            PyExc_ValueError, "a UNIX socket addr was expected");
+        return NULL;
+    }
+
+#ifdef __linux__
+    int addrlen = sizeof (struct sockaddr_un);
+    size_t linuxaddrlen = addrlen - offsetof(struct sockaddr_un, sun_path);
+    if (linuxaddrlen > 0 && addr->sun_path[0] == 0) {
+        return PyBytes_FromStringAndSize(addr->sun_path, linuxaddrlen);
+    }
+    else
+#endif /* linux */
+    {
+        /* regular NULL-terminated string */
+        return PyUnicode_DecodeFSDefault(addr->sun_path);
+    }
+}
+
+
 #if PY_VERSION_HEX < 0x03070000
 
 PyObject * Context_CopyCurrent(void) {

uvloop/includes/system.pxd

@@ -80,3 +80,4 @@ cdef extern from "includes/compat.h" nogil:
 
     int EPOLL_CTL_DEL
     int epoll_ctl(int epfd, int op, int fd, epoll_event *event)
+    object MakeUnixSockPyAddr(sockaddr_un *addr)