Dispatch updates to event handlers

telethon/client/telegrambaseclient.py

@@ -12,7 +12,6 @@ from ..network.mtprotostate import MTProtoState
 from ..sessions import Session, SQLiteSession
 from ..tl import TLObject, functions
 from ..tl.all_tlobjects import LAYER
-from ..update_state import UpdateState
 
 DEFAULT_DC_ID = 4
 DEFAULT_IPV4_IP = '149.154.167.51'
@@ -171,17 +170,14 @@ class TelegramBaseClient(abc.ABC):
         self._connection = connection
         self._sender = MTProtoSender(
             state, connection,
-            first_query=self._init_with(functions.help.GetConfigRequest())
+            first_query=self._init_with(functions.help.GetConfigRequest()),
+            update_callback=self._handle_update
         )
 
         # Cache :tl:`ExportedAuthorization` as ``dc_id: MTProtoState``
         # to easily import them when getting an exported sender.
         self._exported_auths = {}
 
-        # This member will process updates if enabled.
-        # One may change self.updates.enabled at any later point.
-        self.updates = UpdateState()
-
         # Save whether the user is authorized here (a.k.a. logged in)
         self._authorized = None  # None = We don't know yet
 
@@ -367,4 +363,8 @@ class TelegramBaseClient(abc.ABC):
                       'use client(...) instead')
         return await self(*args, **kwargs)
 
+    @abc.abstractmethod
+    def _handle_update(self, update):
+        raise NotImplementedError
+
     # endregion

telethon/client/updates.py

@@ -1,10 +1,10 @@
+import asyncio
+import logging
 import warnings
 
 from .users import UserMethods
+from .. import events, utils
 from ..tl import types, functions
-from .. import events
-
-import logging
 
 __log__ = logging.getLogger(__name__)
 
@@ -44,7 +44,6 @@ class UpdateMethods(UserMethods):
                 :tl:`Update` objects with no further processing) will
                 be passed instead.
         """
-        self.updates.handler = self._on_handler
         if isinstance(event, type):
             event = event()
         elif not event:
@@ -124,7 +123,7 @@ class UpdateMethods(UserMethods):
                         # infinite loop here (so check against old pts to stop)
                         break
 
-                    self.updates.process(types.Updates(
+                    self._handle_update(types.Updates(
                         users=d.users,
                         chats=d.chats,
                         date=state.date,
@@ -144,8 +143,12 @@ class UpdateMethods(UserMethods):
 
     # region Private methods
 
-    async def _on_handler(self, update):
+    def _handle_update(self, update):
+        asyncio.ensure_future(self._dispatch_update(update))
+
+    async def _dispatch_update(self, update):
         if self._events_pending_resolve:
+            # TODO Add lock not to resolve them twice
             for event in self._events_pending_resolve:
                 await event.resolve(self)
             self._events_pending_resolve.clear()