schedule webhook data forwarding

tasks/__init__.py

@@ -1,7 +1,9 @@
 from .channel_history import ChannelHistoryReadTask
 from .mtproto_task_abstraction import MtProtoTask
+from .webhook import WebHookDataForward
 
 __all__ = [
     "ChannelHistoryReadTask",
-    "MtProtoTask"
+    "MtProtoTask",
+    "WebHookDataForward"
 ]

tasks/channel_history.py

@@ -1,9 +1,9 @@
 import pyrogram
-import aiohttp
 import json
 import typing
 
 from .mtproto_task_abstraction import MtProtoTask
+from .webhook import WebHookDataForward
 
 from pyrogram.api.types import ChannelMessagesFilterEmpty
 from pyrogram.api.types.updates import ChannelDifference, ChannelDifferenceTooLong, ChannelDifferenceEmpty
@@ -61,42 +61,40 @@ class JsonSerializer:
 
 
 class ChannelHistoryReadTask(MtProtoTask):
-    channel: pyrogram.Chat
+    _channel: pyrogram.Chat
-    client: pyrogram.Client
+    _client: pyrogram.Client
-    pts: int
+    _pts: int
-    webhook: str
+    _webhook: str
-    http: aiohttp.ClientSession
 
     def setup(self, client: pyrogram.Client, channel: pyrogram.Chat, webhook: str):
-        self.client = client
+        self._pts = False
-        self.channel = channel
-        self.pts = False
-        self.webhook = webhook
-        self.http = aiohttp.ClientSession()
 
-    async def mt_process(self) -> typing.Union[bool, int]:
+        self._client = client
-        response = await self.client.send(
+        self._channel = channel
+        self._webhook = webhook
+
+    async def process(self) -> typing.Union[bool, int]:
+        response = await self._client.send(
             GetChannelDifference(
-                channel=self.channel,
+                channel=self._channel,
                 filter=ChannelMessagesFilterEmpty(),
-                pts=self.pts if self.pts else 0xFFFFFFF,
+                pts=self._pts if self._pts else 0xFFFFFFF,
                 limit=0xFFFFFFF,
                 force=True
             )
         )
 
         if isinstance(response, ChannelDifference):
-            self.pts = response.pts
+            self._pts = response.pts
 
             users = {i.id: i for i in response.users}
             chats = {i.id: i for i in response.chats}
 
             for message in response.new_messages:
-                message = await MessagePyrogram._parse(self.client, message, users, chats)
+                message = await MessagePyrogram._parse(self._client, message, users, chats)
-                message = {"update_id": 1, "message": message}
 
                 data = json.dumps(
-                    message,
+                    {"update_id": 1, "message": message},
                     default=JsonSerializer.default,
                     ensure_ascii=True,
                     allow_nan=False,
@@ -104,16 +102,9 @@ class ChannelHistoryReadTask(MtProtoTask):
                     sort_keys=False
                 )
 
-                result = await self.http.post(
+                forwarder = WebHookDataForward()
-                    self.webhook,
+                forwarder.setup(self._webhook, data)
-                    data=data,
+                await self.future(forwarder)
-                    headers=[
-                        ("Content-Type", "application/json")
-                    ]
-                )
-
-                await result.read()
-                result.close()
 
             if not response.final:
                 return 1
@@ -121,7 +112,7 @@ class ChannelHistoryReadTask(MtProtoTask):
             return response.timeout
 
         if isinstance(response, ChannelDifferenceEmpty):
-            self.pts = response.pts
+            self._pts = response.pts
             return response.timeout
 
         if isinstance(response, ChannelDifferenceTooLong):

tasks/mtproto_task_abstraction.py

@@ -12,16 +12,15 @@ class MtProtoTask(AsyncTask, abc.ABC):
         raise NotImplementedError
 
     @abc.abstractmethod
-    async def mt_process(self) -> typing.Union[bool, int]:
+    async def process(self) -> typing.Union[bool, int]:
         raise NotImplementedError
 
-    async def process(self) -> int:
+    async def _process(self) -> typing.Union[bool, int]:
         try:
-
+            return await self.process() * 1e9
-            return await self.mt_process()
 
         except FloodWait as error:
-            return int(error.MESSAGE.split("_")[-1])
+            return int(error.MESSAGE.split("_")[-1]) * 1e9
 
         except (ChannelInvalid, ChannelPrivate, Unauthorized):
             return False

tasks/webhook.py

@@ -0,0 +1,24 @@
+import typing
+import aiohttp
+
+from async_worker import OneLoopAsyncTask
+
+
+HEADERS = [("Content-Type", "application/json")]
+
+
+class WebHookDataForward(OneLoopAsyncTask):
+    _webhook: str
+    _data: typing.Union[str, bytes]
+    _http: aiohttp.ClientSession
+
+    async def process(self) -> typing.NoReturn:
+        res = await self._http.post(self._webhook, data=self._data, headers=HEADERS)
+        await res.read()
+        res.close()
+
+    def setup(self, webhook: str, data: typing.Union[str, bytes]):
+        self._http = aiohttp.ClientSession()
+
+        self._webhook = webhook
+        self._data = data