[Python-checkins] cpython (merge 3.4 -> default): (Merge 3.4) asyncion, Tulip issue 181: BaseEventLoop.create_datagram_endpoint()

[victor.stinner]

http://hg.python.org/cpython/rev/50df498725f1
changeset:   91615:50df498725f1
parent:      91613:2c9d5f32f6c5
parent:      91614:afdbc02e4197
user:        Victor Stinner <victor.stinner at gmail.com>
date:        Tue Jul 08 23:58:25 2014 +0200
summary:
  (Merge 3.4) asyncion, Tulip issue 181: BaseEventLoop.create_datagram_endpoint()
now waits until protocol.connection_made() has been called. Document also why
transport constructors use a waiter.

files:
  Lib/asyncio/base_events.py           |   7 +++++--
  Lib/asyncio/proactor_events.py       |   1 +
  Lib/asyncio/selector_events.py       |  13 ++++++++++---
  Lib/asyncio/unix_events.py           |   2 ++
  Lib/test/test_asyncio/test_events.py |  10 ++++++++++
  5 files changed, 28 insertions(+), 5 deletions(-)


diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -169,7 +169,7 @@
         raise NotImplementedError
 
     def _make_datagram_transport(self, sock, protocol,
-                                 address=None, extra=None):
+                                 address=None, waiter=None, extra=None):
         """Create datagram transport."""
         raise NotImplementedError
 
@@ -605,7 +605,10 @@
             raise exceptions[0]
 
         protocol = protocol_factory()
-        transport = self._make_datagram_transport(sock, protocol, r_addr)
+        waiter = futures.Future(loop=self)
+        transport = self._make_datagram_transport(sock, protocol, r_addr,
+                                                  waiter)
+        yield from waiter
         return transport, protocol
 
     @coroutine
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -38,6 +38,7 @@
             self._server.attach(self)
         self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
+            # wait until protocol.connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
 
     def _set_extra(self, sock):
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -51,8 +51,9 @@
             server_side, server_hostname, extra, server)
 
     def _make_datagram_transport(self, sock, protocol,
-                                 address=None, extra=None):
-        return _SelectorDatagramTransport(self, sock, protocol, address, extra)
+                                 address=None, waiter=None, extra=None):
+        return _SelectorDatagramTransport(self, sock, protocol,
+                                          address, waiter, extra)
 
     def close(self):
         if self.is_closed():
@@ -481,6 +482,7 @@
         self._loop.add_reader(self._sock_fd, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
+            # wait until protocol.connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
 
     def pause_reading(self):
@@ -690,6 +692,7 @@
         self._loop.add_reader(self._sock_fd, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
         if self._waiter is not None:
+            # wait until protocol.connection_made() has been called
             self._loop.call_soon(self._waiter._set_result_unless_cancelled,
                                  None)
 
@@ -806,11 +809,15 @@
 
     _buffer_factory = collections.deque
 
-    def __init__(self, loop, sock, protocol, address=None, extra=None):
+    def __init__(self, loop, sock, protocol, address=None,
+                 waiter=None, extra=None):
         super().__init__(loop, sock, protocol, extra)
         self._address = address
         self._loop.add_reader(self._sock_fd, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
+        if waiter is not None:
+            # wait until protocol.connection_made() has been called
+            self._loop.call_soon(waiter._set_result_unless_cancelled, None)
 
     def get_write_buffer_size(self):
         return sum(len(data) for data, _ in self._buffer)
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -269,6 +269,7 @@
         self._loop.add_reader(self._fileno, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
+            # wait until protocol.connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
 
     def _read_ready(self):
@@ -353,6 +354,7 @@
 
         self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
+            # wait until protocol.connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
 
     def get_write_buffer_size(self):
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -522,6 +522,7 @@
         tr, pr = self.loop.run_until_complete(connection_fut)
         self.assertIsInstance(tr, asyncio.Transport)
         self.assertIsInstance(pr, asyncio.Protocol)
+        self.assertIs(pr.transport, tr)
         if check_sockname:
             self.assertIsNotNone(tr.get_extra_info('sockname'))
         self.loop.run_until_complete(pr.done)
@@ -1045,12 +1046,21 @@
         s_transport, server = self.loop.run_until_complete(coro)
         host, port = s_transport.get_extra_info('sockname')
 
+        self.assertIsInstance(s_transport, asyncio.Transport)
+        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),
             remote_addr=(host, port))
         transport, client = self.loop.run_until_complete(coro)
 
+        self.assertIsInstance(transport, asyncio.Transport)
+        self.assertIsInstance(client, MyDatagramProto)
         self.assertEqual('INITIALIZED', client.state)
+        self.assertIs(client.transport, transport)
+
         transport.sendto(b'xxx')
         test_utils.run_until(self.loop, lambda: server.nbytes)
         self.assertEqual(3, server.nbytes)

-- 
Repository URL: http://hg.python.org/cpython