(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.
This commit is contained in:
commit
b44fc32661
|
@ -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
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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)
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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)
|
||||
|
|
Loading…
Reference in New Issue