diff options
| -rw-r--r-- | Lib/asyncio/base_events.py | 7 | ||||
| -rw-r--r-- | Lib/asyncio/proactor_events.py | 1 | ||||
| -rw-r--r-- | Lib/asyncio/selector_events.py | 13 | ||||
| -rw-r--r-- | Lib/asyncio/unix_events.py | 2 | ||||
| -rw-r--r-- | 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 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) |