LonamiWebs/Telethon
1
1
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2024-11-25 02:43:45 +03:00
Code Issues Packages Projects Releases Wiki Activity

Propagate errors at the connection level

Browse Source
This commit is contained in:
Lonami Exo 2023-03-12 17:43:19 +01:00
parent f3414d134a
commit acd3407418
1 changed files with 25 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
telethon/network/connection/connection.py
View File

@ -296,8 +296,10 @@ class Connection(abc.ABC):
This method returns a coroutine.
"""
while self._connected:
result = await self._recv_queue.get()
if result: # None = sentinel value = keep trying
result, err = await self._recv_queue.get()
if err:
raise err
if result:
return result
raise ConnectionError('Not connected')
@ -324,37 +326,29 @@ class Connection(abc.ABC):
"""
This loop is constantly putting items on the queue as they're read.
"""
try:
while self._connected:
try:
data = await self._recv()
except asyncio.CancelledError:
break
except Exception as e:
if isinstance(e, (IOError, asyncio.IncompleteReadError)):
msg = 'The server closed the connection'
self._log.info(msg)
elif isinstance(e, InvalidChecksumError):
msg = 'The server response had an invalid checksum'
self._log.info(msg)
elif isinstance(e, InvalidBufferError):
msg = 'The server response had an invalid buffer'
self._log.error(msg)
except (IOError, asyncio.IncompleteReadError) as e:
self._log.warning('Server closed the connection: %s', e)
await self._recv_queue.put((None, e))
except InvalidChecksumError as e:
self._log.warning('Server response had invalid checksum: %s', e)
await self._recv_queue.put((None, e))
except InvalidBufferError as e:
self._log.warning('Server response had invalid buffer: %s', e)
await self._recv_queue.put((None, e))
except Exception:
self._log.exception('Unexpected exception in the receive loop')
await self._recv_queue.put((None, e))
else:
msg = 'Unexpected exception in the receive loop'
self._log.exception(msg)
await self._recv_queue.put((data, None))
except asyncio.CancelledError:
pass
finally:
await self.disconnect()
# Add a sentinel value to unstuck recv
if self._recv_queue.empty():
self._recv_queue.put_nowait(None)
break
try:
await self._recv_queue.put(data)
except asyncio.CancelledError:
break
def _init_conn(self):
"""