Fix event loop not being passed into many asyncio calls

telethon/client/telegrambaseclient.py

@@ -244,7 +244,7 @@ class TelegramBaseClient(abc.ABC):
         # being ``n`` the amount of borrows a given sender has; once ``n``
         # reaches ``0`` it should be disconnected and removed.
         self._borrowed_senders = {}
-        self._borrow_sender_lock = asyncio.Lock()
+        self._borrow_sender_lock = asyncio.Lock(loop=self._loop)
 
         # Save whether the user is authorized here (a.k.a. logged in)
         self._authorized = None  # None = We don't know yet
@@ -258,8 +258,8 @@ class TelegramBaseClient(abc.ABC):
         self._channel_pts = {}
 
         if sequential_updates:
-            self._updates_queue = asyncio.Queue()
+            self._updates_queue = asyncio.Queue(loop=self._loop)
-            self._dispatching_updates_queue = asyncio.Event()
+            self._dispatching_updates_queue = asyncio.Event(loop=self._loop)
         else:
             self._updates_queue = None
             self._dispatching_updates_queue = None

telethon/events/inlinequery.py

@@ -162,8 +162,10 @@ class InlineQuery(EventBuilder):
             if self._answered:
                 return
 
-            results = [self._as_awaitable(x) for x in results]
+            results = [self._as_awaitable(x, self._client.loop)
-            done, _ = await asyncio.wait(results)
+                       for x in results]
+
+            done, _ = await asyncio.wait(results, loop=self._client.loop)
             results = [x.result() for x in done]
 
             if switch_pm:
@@ -181,10 +183,10 @@ class InlineQuery(EventBuilder):
             )
 
         @staticmethod
-        def _as_awaitable(obj):
+        def _as_awaitable(obj, loop):
             if inspect.isawaitable(obj):
                 return obj
 
-            f = asyncio.Future()
+            f = asyncio.Future(loop=loop)
             f.set_result(obj)
             return f

telethon/network/mtprotosender.py

@@ -205,14 +205,16 @@ class MTProtoSender:
             result = []
             after = None
             for r in request:
-                message = self.state.create_message(r, after=after)
+                message = self.state.create_message(
+                    r, loop=self._loop, after=after)
+
                 self._pending_messages[message.msg_id] = message
                 self._send_queue.put_nowait(message)
                 result.append(message.future)
                 after = ordered and message
             return result
         else:
-            message = self.state.create_message(request)
+            message = self.state.create_message(request, loop=self._loop)
             self._pending_messages[message.msg_id] = message
             self._send_queue.put_nowait(message)
             return message.future
@@ -280,7 +282,7 @@ class MTProtoSender:
 
         # First connection or manual reconnection after a failure
         if self._disconnected is None or self._disconnected.done():
-            self._disconnected = asyncio.Future()
+            self._disconnected = asyncio.Future(loop=self._loop)
         __log__.info('Connection to {} complete!'.format(self._ip))
 
     async def _reconnect(self):
@@ -352,7 +354,7 @@ class MTProtoSender:
         while self._user_connected and not self._reconnecting:
             if self._pending_ack:
                 self._last_ack = self.state.create_message(
-                    MsgsAck(list(self._pending_ack))
+                    MsgsAck(list(self._pending_ack)), loop=self._loop
                 )
                 self._send_queue.put_nowait(self._last_ack)
                 self._pending_ack.clear()
@@ -365,7 +367,9 @@ class MTProtoSender:
                     continue
 
             if isinstance(messages, list):
-                message = self.state.create_message(MessageContainer(messages))
+                message = self.state.create_message(
+                    MessageContainer(messages), loop=self._loop)
+
                 self._pending_messages[message.msg_id] = message
                 self._pending_containers.append(message)
             else:
@@ -394,7 +398,7 @@ class MTProtoSender:
                         __log__.warning('OSError while sending %s', e)
                     else:
                         __log__.exception('Unhandled exception while receiving')
-                        await asyncio.sleep(1)
+                        await asyncio.sleep(1, loop=self._loop)
 
                     self._start_reconnect()
                     break
@@ -433,7 +437,7 @@ class MTProtoSender:
                     __log__.warning('OSError while receiving %s', e)
                 else:
                     __log__.exception('Unhandled exception while receiving')
-                    await asyncio.sleep(1)
+                    await asyncio.sleep(1, loop=self._loop)
 
                 self._start_reconnect()
                 break
@@ -471,7 +475,7 @@ class MTProtoSender:
                 return
             except Exception as e:
                 __log__.exception('Unhandled exception while unpacking %s',e)
-                await asyncio.sleep(1)
+                await asyncio.sleep(1, loop=self._loop)
             else:
                 try:
                     await self._process_message(message)
@@ -480,7 +484,7 @@ class MTProtoSender:
                 except Exception as e:
                     __log__.exception('Unhandled exception while '
                                       'processing %s', message)
-                    await asyncio.sleep(1)
+                    await asyncio.sleep(1, loop=self._loop)
 
     # Response Handlers
 
@@ -525,7 +529,7 @@ class MTProtoSender:
         if rpc_result.error:
             error = rpc_message_to_error(rpc_result.error)
             self._send_queue.put_nowait(self.state.create_message(
-                MsgsAck([message.msg_id])
+                MsgsAck([message.msg_id]), loop=self._loop
             ))
 
             if not message.future.cancelled():

telethon/network/mtprotostate.py

@@ -37,7 +37,7 @@ class MTProtoState:
         self._sequence = 0
         self._last_msg_id = 0
 
-    def create_message(self, obj, after=None):
+    def create_message(self, obj, *, loop, after=None):
         """
         Creates a new `telethon.tl.tl_message.TLMessage` from
         the given `telethon.tl.tlobject.TLObject` instance.
@@ -47,7 +47,8 @@ class MTProtoState:
             seq_no=self._get_seq_no(isinstance(obj, TLRequest)),
             obj=obj,
             after_id=after.msg_id if after else None,
-            out=True  # Pre-convert the request into bytes
+            out=True,  # Pre-convert the request into bytes
+            loop=loop
         )
 
     def update_message_id(self, message):
@@ -135,7 +136,7 @@ class MTProtoState:
         # reader isn't used for anything else after this, it's unnecessary.
         obj = reader.tgread_object()
 
-        return TLMessage(remote_msg_id, remote_sequence, obj)
+        return TLMessage(remote_msg_id, remote_sequence, obj, loop=None)
 
     def _get_new_msg_id(self):
         """

telethon/tl/core/messagecontainer.py

@@ -43,5 +43,5 @@ class MessageContainer(TLObject):
             before = reader.tell_position()
             obj = reader.tgread_object()  # May over-read e.g. RpcResult
             reader.set_position(before + length)
-            messages.append(TLMessage(msg_id, seq_no, obj))
+            messages.append(TLMessage(msg_id, seq_no, obj, loop=None))
         return MessageContainer(messages)

telethon/tl/core/tlmessage.py

@@ -24,10 +24,13 @@ class TLMessage(TLObject):
     sent `TLMessage`, and this result can be represented as a `Future`
     that will eventually be set with either a result, error or cancelled.
     """
-    def __init__(self, msg_id, seq_no, obj, out=False, after_id=0):
+    def __init__(self, msg_id, seq_no, obj, *, loop, out=False, after_id=0):
         self.obj = obj
         self.container_msg_id = None
-        self.future = asyncio.Future()
+
+        # If no loop is given then it is an incoming message.
+        # Only outgoing messages need the future to await them.
+        self.future = asyncio.Future(loop=loop) if loop else None
 
         # After which message ID this one should run. We do this so
         # InvokeAfterMsgRequest is transparent to the user and we can

telethon/tl/custom/conversation.py

@@ -181,7 +181,7 @@ class Conversation(ChatGetter):
                 return incoming
 
         # Otherwise the next incoming response will be the one to use
-        future = asyncio.Future()
+        future = asyncio.Future(loop=self._client.loop)
         pending[target_id] = future
         return self._get_result(future, start_time, timeout)
 
@@ -209,7 +209,7 @@ class Conversation(ChatGetter):
             return earliest_edit
 
         # Otherwise the next incoming response will be the one to use
-        future = asyncio.Future()
+        future = asyncio.Future(loop=self._client.loop)
         self._pending_edits[target_id] = future
         return await self._get_result(future, start_time, timeout)
 
@@ -220,7 +220,7 @@ class Conversation(ChatGetter):
         will also trigger even without a response.
         """
         start_time = time.time()
-        future = asyncio.Future()
+        future = asyncio.Future(loop=self._client.loop)
         target_id = self._get_message_id(message)
 
         if self._last_read is None:
@@ -265,7 +265,7 @@ class Conversation(ChatGetter):
         counter = Conversation._custom_counter
         Conversation._custom_counter += 1
 
-        future = asyncio.Future()
+        future = asyncio.Future(loop=self._client.loop)
         async def result():
             try:
                 return await self._get_result(future, start_time, timeout)