summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorVictor Stinner <victor.stinner@gmail.com>2014-07-08 21:57:31 (GMT)
committerVictor Stinner <victor.stinner@gmail.com>2014-07-08 21:57:31 (GMT)
commitbfff45d611f3435ed4b713124f65fba8e961ff03 (patch)
tree752da659a58b6ac8178ff41a9d890cd83f2e7177
parentaea82293be0a56937fe807a0d2a44c7db392f155 (diff)
downloadcpython-bfff45d611f3435ed4b713124f65fba8e961ff03.zip
cpython-bfff45d611f3435ed4b713124f65fba8e961ff03.tar.gz
cpython-bfff45d611f3435ed4b713124f65fba8e961ff03.tar.bz2
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.
-rw-r--r--Lib/asyncio/base_events.py7
-rw-r--r--Lib/asyncio/proactor_events.py1
-rw-r--r--Lib/asyncio/selector_events.py13
-rw-r--r--Lib/asyncio/unix_events.py2
-rw-r--r--Lib/test/test_asyncio/test_events.py10
5 files changed, 28 insertions, 5 deletions
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index 52c5517..833f81d 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -169,7 +169,7 @@ class BaseEventLoop(events.AbstractEventLoop):
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 @@ class BaseEventLoop(events.AbstractEventLoop):
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
index a80876f..fa24795 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -38,6 +38,7 @@ class _ProactorBasePipeTransport(transports._FlowControlMixin,
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
index 2a17034..7b364ad 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -51,8 +51,9 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
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 @@ class _SelectorSocketTransport(_SelectorTransport):
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 @@ class _SelectorSslTransport(_SelectorTransport):
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 @@ class _SelectorDatagramTransport(_SelectorTransport):
_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
index 535ea22..764e719 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -269,6 +269,7 @@ class _UnixReadPipeTransport(transports.ReadTransport):
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 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
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
index b89416f..e5c5729 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -522,6 +522,7 @@ class EventLoopTestsMixin:
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 @@ class EventLoopTestsMixin:
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)