mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2025-06-05 06:03:30 +03:00
Handle initial connection if network is down correctly
This commit is contained in:
parent
2d275989cb
commit
5edc2216c7
|
@ -44,9 +44,14 @@ class Connection(abc.ABC):
|
||||||
Disconnects from the server.
|
Disconnects from the server.
|
||||||
"""
|
"""
|
||||||
self._disconnected.set()
|
self._disconnected.set()
|
||||||
self._send_task.cancel()
|
if self._send_task:
|
||||||
self._recv_task.cancel()
|
self._send_task.cancel()
|
||||||
self._writer.close()
|
|
||||||
|
if self._recv_task:
|
||||||
|
self._recv_task.cancel()
|
||||||
|
|
||||||
|
if self._writer:
|
||||||
|
self._writer.close()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def disconnected(self):
|
def disconnected(self):
|
||||||
|
|
|
@ -126,41 +126,6 @@ class MTProtoSender:
|
||||||
|
|
||||||
await self._disconnect()
|
await self._disconnect()
|
||||||
|
|
||||||
# TODO Move this out of the "Public API" section
|
|
||||||
async def _disconnect(self, error=None):
|
|
||||||
__log__.info('Disconnecting from %s...', self._connection)
|
|
||||||
self._user_connected = False
|
|
||||||
try:
|
|
||||||
__log__.debug('Closing current connection...')
|
|
||||||
self._connection.disconnect()
|
|
||||||
finally:
|
|
||||||
__log__.debug('Cancelling {} pending message(s)...'
|
|
||||||
.format(len(self._pending_state)))
|
|
||||||
for state in self._pending_state.values():
|
|
||||||
if error and not state.future.done():
|
|
||||||
state.future.set_exception(error)
|
|
||||||
else:
|
|
||||||
state.future.cancel()
|
|
||||||
|
|
||||||
self._pending_state.clear()
|
|
||||||
self._pending_ack.clear()
|
|
||||||
self._last_ack = None
|
|
||||||
|
|
||||||
if self._send_loop_handle:
|
|
||||||
__log__.debug('Cancelling the send loop...')
|
|
||||||
self._send_loop_handle.cancel()
|
|
||||||
|
|
||||||
if self._recv_loop_handle:
|
|
||||||
__log__.debug('Cancelling the receive loop...')
|
|
||||||
self._recv_loop_handle.cancel()
|
|
||||||
|
|
||||||
__log__.info('Disconnection from %s complete!', self._connection)
|
|
||||||
if self._disconnected and not self._disconnected.done():
|
|
||||||
if error:
|
|
||||||
self._disconnected.set_exception(error)
|
|
||||||
else:
|
|
||||||
self._disconnected.set_result(None)
|
|
||||||
|
|
||||||
def send(self, request, ordered=False):
|
def send(self, request, ordered=False):
|
||||||
"""
|
"""
|
||||||
This method enqueues the given request to be sent. Its send
|
This method enqueues the given request to be sent. Its send
|
||||||
|
@ -224,12 +189,14 @@ class MTProtoSender:
|
||||||
authorization key if necessary, and starting the send and
|
authorization key if necessary, and starting the send and
|
||||||
receive loops.
|
receive loops.
|
||||||
"""
|
"""
|
||||||
|
# TODO With ``asyncio.open_connection``, no timeout occurs
|
||||||
|
# However, these are probably desirable in some circumstances.
|
||||||
__log__.info('Connecting to %s...', self._connection)
|
__log__.info('Connecting to %s...', self._connection)
|
||||||
for retry in range(1, self._retries + 1):
|
for retry in range(1, self._retries + 1):
|
||||||
try:
|
try:
|
||||||
__log__.debug('Connection attempt {}...'.format(retry))
|
__log__.debug('Connection attempt {}...'.format(retry))
|
||||||
await self._connection.connect()
|
await self._connection.connect()
|
||||||
except (asyncio.TimeoutError, OSError) as e:
|
except OSError as e:
|
||||||
__log__.warning('Attempt {} at connecting failed: {}: {}'
|
__log__.warning('Attempt {} at connecting failed: {}: {}'
|
||||||
.format(retry, type(e).__name__, e))
|
.format(retry, type(e).__name__, e))
|
||||||
else:
|
else:
|
||||||
|
@ -274,6 +241,40 @@ class MTProtoSender:
|
||||||
self._disconnected = self._loop.create_future()
|
self._disconnected = self._loop.create_future()
|
||||||
__log__.info('Connection to %s complete!', self._connection)
|
__log__.info('Connection to %s complete!', self._connection)
|
||||||
|
|
||||||
|
async def _disconnect(self, error=None):
|
||||||
|
__log__.info('Disconnecting from %s...', self._connection)
|
||||||
|
self._user_connected = False
|
||||||
|
try:
|
||||||
|
__log__.debug('Closing current connection...')
|
||||||
|
self._connection.disconnect()
|
||||||
|
finally:
|
||||||
|
__log__.debug('Cancelling {} pending message(s)...'
|
||||||
|
.format(len(self._pending_state)))
|
||||||
|
for state in self._pending_state.values():
|
||||||
|
if error and not state.future.done():
|
||||||
|
state.future.set_exception(error)
|
||||||
|
else:
|
||||||
|
state.future.cancel()
|
||||||
|
|
||||||
|
self._pending_state.clear()
|
||||||
|
self._pending_ack.clear()
|
||||||
|
self._last_ack = None
|
||||||
|
|
||||||
|
if self._send_loop_handle:
|
||||||
|
__log__.debug('Cancelling the send loop...')
|
||||||
|
self._send_loop_handle.cancel()
|
||||||
|
|
||||||
|
if self._recv_loop_handle:
|
||||||
|
__log__.debug('Cancelling the receive loop...')
|
||||||
|
self._recv_loop_handle.cancel()
|
||||||
|
|
||||||
|
__log__.info('Disconnection from %s complete!', self._connection)
|
||||||
|
if self._disconnected and not self._disconnected.done():
|
||||||
|
if error:
|
||||||
|
self._disconnected.set_exception(error)
|
||||||
|
else:
|
||||||
|
self._disconnected.set_result(None)
|
||||||
|
|
||||||
async def _reconnect(self):
|
async def _reconnect(self):
|
||||||
"""
|
"""
|
||||||
Cleanly disconnects and then reconnects.
|
Cleanly disconnects and then reconnects.
|
||||||
|
@ -340,7 +341,7 @@ class MTProtoSender:
|
||||||
|
|
||||||
# TODO Debug logs to notify which messages are being sent
|
# TODO Debug logs to notify which messages are being sent
|
||||||
# TODO Try sending them while no future was cancelled?
|
# TODO Try sending them while no future was cancelled?
|
||||||
# TODO Handle timeout, cancelled, arbitrary errors
|
# TODO Handle cancelled?, arbitrary errors
|
||||||
await self._connection.send(state_list)
|
await self._connection.send(state_list)
|
||||||
for state in state_list:
|
for state in state_list:
|
||||||
if not isinstance(state, list):
|
if not isinstance(state, list):
|
||||||
|
@ -411,10 +412,6 @@ class MTProtoSender:
|
||||||
if not state.future.cancelled():
|
if not state.future.cancelled():
|
||||||
state.future.set_exception(error)
|
state.future.set_exception(error)
|
||||||
else:
|
else:
|
||||||
# TODO Would be nice to avoid accessing a per-obj read_result
|
|
||||||
# Instead have a variable that indicated how the result should
|
|
||||||
# be read (an enum) and dispatch to read the result, mostly
|
|
||||||
# always it's just a normal TLObject.
|
|
||||||
with BinaryReader(rpc_result.body) as reader:
|
with BinaryReader(rpc_result.body) as reader:
|
||||||
result = state.request.read_result(reader)
|
result = state.request.read_result(reader)
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue
Block a user