Implement loop.create_datagram_endpoint

Author: 1st1

tests/test_udp.py

@@ -0,0 +1,128 @@
+import asyncio
+import logging
+import socket
+import uvloop
+import ssl
+import warnings
+
+from asyncio import test_utils
+from uvloop import _testbase as tb
+
+
+class MyDatagramProto(asyncio.DatagramProtocol):
+    done = None
+
+    def __init__(self, loop=None):
+        self.state = 'INITIAL'
+        self.nbytes = 0
+        if loop is not None:
+            self.done = asyncio.Future(loop=loop)
+
+    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.nbytes += len(data)
+
+    def error_received(self, exc):
+        assert self.state == 'INITIALIZED', self.state
+        raise exc
+
+    def connection_lost(self, exc):
+        assert self.state == 'INITIALIZED', self.state
+        self.state = 'CLOSED'
+        if self.done:
+            self.done.set_result(None)
+
+
+class _TestUDP:
+    def test_create_datagram_endpoint_addrs(self):
+        class TestMyDatagramProto(MyDatagramProto):
+            def __init__(inner_self):
+                super().__init__(loop=self.loop)
+
+            def datagram_received(self, data, addr):
+                super().datagram_received(data, addr)
+                self.transport.sendto(b'resp:' + data, addr)
+
+        for lc in (('127.0.0.1', 0), None):
+            if lc is None and not isinstance(self.loop, uvloop._Loop):
+                # TODO This looks like a bug in asyncio -- if no local_addr
+                # and no remote_addr are specified, the connection
+                # that asyncio creates is not bound anywhere.
+                return
+
+            with self.subTest(local_addr=lc):
+                coro = self.loop.create_datagram_endpoint(
+                    TestMyDatagramProto, local_addr=lc, family=socket.AF_INET)
+                s_transport, server = self.loop.run_until_complete(coro)
+                host, port = s_transport.get_extra_info('sockname')
+
+                self.assertIsInstance(server, TestMyDatagramProto)
+                self.assertEqual('INITIALIZED', server.state)
+                self.assertIs(server.transport, s_transport)
+
+                coro = self.loop.create_datagram_endpoint(
+                    lambda: MyDatagramProto(loop=self.loop),
+                    family=socket.AF_INET,
+                    remote_addr=None if lc is None else (host, port))
+                transport, client = self.loop.run_until_complete(coro)
+
+                self.assertIsInstance(client, MyDatagramProto)
+                self.assertEqual('INITIALIZED', client.state)
+                self.assertIs(client.transport, transport)
+
+                transport.sendto(b'xxx', (host, port) if lc is None else None)
+                test_utils.run_until(self.loop, lambda: server.nbytes)
+                self.assertEqual(3, server.nbytes)
+                test_utils.run_until(self.loop, lambda: client.nbytes)
+
+                # received
+                self.assertEqual(8, client.nbytes)
+
+                # extra info is available
+                self.assertIsNotNone(transport.get_extra_info('sockname'))
+
+                # close connection
+                transport.close()
+                self.loop.run_until_complete(client.done)
+                self.assertEqual('CLOSED', client.state)
+                server.transport.close()
+                self.loop.run_until_complete(server.done)
+
+    def test_create_datagram_endpoint_sock(self):
+        sock = None
+        local_address = ('127.0.0.1', 0)
+        infos = self.loop.run_until_complete(
+            self.loop.getaddrinfo(
+                *local_address, type=socket.SOCK_DGRAM))
+        for family, type, proto, cname, address in infos:
+            try:
+                sock = socket.socket(family=family, type=type, proto=proto)
+                sock.setblocking(False)
+                sock.bind(address)
+            except:
+                pass
+            else:
+                break
+        else:
+            assert False, 'Can not create socket.'
+
+        with sock:
+            f = self.loop.create_connection(
+                lambda: MyDatagramProto(loop=self.loop), sock=sock)
+            tr, pr = self.loop.run_until_complete(f)
+            self.assertIsInstance(pr, MyDatagramProto)
+            tr.close()
+            self.loop.run_until_complete(pr.done)
+
+
+class Test_UV_UDP(_TestUDP, tb.UVTestCase):
+    pass
+
+
+class Test_AIO_UDP(_TestUDP, tb.AIOTestCase):
+    pass

uvloop/dns.pyx

@@ -39,6 +39,62 @@ cdef __convert_sockaddr_to_pyaddr(system.sockaddr* addr):
     raise RuntimeError("cannot convert sockaddr into Python object")
 
 
+cdef __convert_pyaddr_to_sockaddr(int family, object addr,
+                                  system.sockaddr* res):
+    cdef:
+        int err
+        int port
+        int addr_len
+        int scope_id = 0
+        int flowinfo = 0
+
+    if family == uv.AF_INET:
+        if not isinstance(addr, tuple):
+            raise TypeError('AF_INET address must be tuple')
+        if len(addr) != 2:
+            raise ValueError('AF_INET address must be tuple of (host, port)')
+        host, port = addr
+        if isinstance(host, str):
+            host = host.encode()
+        if not isinstance(host, (bytes, bytearray)):
+            raise TypeError('host must be a string or bytes object')
+
+        err = uv.uv_ip4_addr(host, port, <system.sockaddr_in*>res)
+        if err < 0:
+            raise convert_error(err)
+
+    elif family == uv.AF_INET6:
+        if not isinstance(addr, tuple):
+            raise TypeError('AF_INET6 address must be tuple')
+
+        addr_len = len(addr)
+        if addr_len < 2 or addr_len > 4:
+            raise ValueError(
+                'AF_INET6 must be a tuple of 2-4 parameters: '
+                '(host, port, flowinfo?, scope_id?)')
+
+        host = addr[0]
+        if isinstance(host, str):
+            host = host.encode()
+
+        port = addr[1]
+        if addr_len > 2:
+            flowinfo = addr[2]
+        if addr_len > 3:
+            scope_id = addr[3]
+
+        err = uv.uv_ip6_addr(host, port, <system.sockaddr_in6*>res)
+        if err < 0:
+            raise convert_error(err)
+
+        (<system.sockaddr_in6*>res).sin6_flowinfo = flowinfo
+        (<system.sockaddr_in6*>res).sin6_scope_id = scope_id
+
+    else:
+        raise ValueError(
+            'epected AF_INET or AF_INET6 family, got {}'.format(family))
+
+
 @cython.freelist(DEFAULT_FREELIST_SIZE)
 cdef class AddrInfo:
     cdef:
@@ -58,6 +114,7 @@ cdef class AddrInfo:
     cdef unpack(self):
         cdef:
             list result = []
+            system.addrinfo *ptr
 
         if self.data is NULL:
             raise RuntimeError('AddrInfo.data is NULL')

uvloop/handles/udp.pxd

@@ -0,0 +1,24 @@
+cdef class UVUDP(UVBaseTransport):
+    cdef:
+        bint __receiving
+        int _family
+
+        bint _address_set
+        system.sockaddr _address
+        object _cached_py_address
+
+    cdef _init(self, Loop loop, unsigned int family)
+
+    cdef _set_remote_address(self, system.sockaddr address)
+
+    cdef _bind(self, system.sockaddr* addr, bint reuse_addr)
+    cdef _open(self, int family, int sockfd)
+    cdef _set_broadcast(self, bint on)
+
+    cdef inline __receiving_started(self)
+    cdef inline __receiving_stopped(self)
+
+    cdef _send(self, object data, object addr)
+
+    cdef _on_receive(self, bytes data, object exc, object addr)
+    cdef _on_sent(self, object exc)