LonamiWebs/Telethon
1
1
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2025-08-06 13:10:22 +03:00
Code Issues Packages Projects Releases Wiki Activity

Fix writing in sender

Browse Source
This commit is contained in:
Jahongir Qurbonov 2024-10-22 14:43:50 +05:00
parent d77bc7abe9
commit a50fc8a375
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
client/src/telethon/_impl/mtsender/sender.py
View File

@ -204,10 +204,11 @@ class Sender:
ticket_number = self._step_counter ticket_number = self._step_counter
async with self.lock: async with self.lock:
self._try_fill_write()
if self._step_counter == ticket_number: if self._step_counter == ticket_number:
# We're the one to drive IO. # We're the one to drive IO.
self._step_counter += 1 self._step_counter += 1
await self._step() await self._wait_response()
# else: different task drove IO. # else: different task drove IO.
def pop_updates(self) -> list[Updates]: def pop_updates(self) -> list[Updates]:
@ -215,15 +216,6 @@ class Sender:
self._updates.clear() self._updates.clear()
return updates return updates
async def _step(self) -> None:
assert self._protocol
if self._protocol.is_closed():
raise ConnectionResetError
self._response_state.clear()
self._try_fill_write()
await self._wait_response()
def _try_fill_write(self) -> None: def _try_fill_write(self) -> None:
for request in self._requests: for request in self._requests:
if isinstance(request.state, NotSerialized): if isinstance(request.state, NotSerialized):
@ -242,6 +234,12 @@ class Sender:
request.state = Sent(request.state.msg_id, container_msg_id) request.state = Sent(request.state.msg_id, container_msg_id)
async def _wait_response(self) -> None: async def _wait_response(self) -> None:
assert self._protocol
if self._protocol.is_closed():
raise ConnectionResetError
self._response_state.clear()
try: try:
async with asyncio.timeout(PING_DELAY): async with asyncio.timeout(PING_DELAY):
await self._response_state.wait() await self._response_state.wait()
@ -264,7 +262,6 @@ class Sender:
self._read_buffer += bytes(n) self._read_buffer += bytes(n)
self._read_buffer_head -= n self._read_buffer_head -= n
self._process_mtp_buffer() self._process_mtp_buffer()
finally:
self._response_state.set() self._response_state.set()
def _on_conn_closed(self) -> None: def _on_conn_closed(self) -> None: