PersonalPublic/mosgortrans
1
0
Fork 0
mirror of https://github.com/Balshgit/mosgortrans.git synced 2025-09-11 13:00:40 +03:00
Code Issues Packages Projects Releases Wiki Activity

add async queue

Browse Source
This commit is contained in:
Dmitry Afanasyev 2022-08-20 03:07:50 +03:00
parent 04d31b0e24
commit cdf2720e57
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
app/main.py
View File

@ -1,5 +1,6 @@
import asyncio import asyncio
import sys import sys
from concurrent.futures import ThreadPoolExecutor
from http import HTTPStatus from http import HTTPStatus
from pathlib import Path from pathlib import Path
@ -22,6 +23,8 @@ from app.settings import (
WEBHOOK_URL, WEBHOOK_URL,
) )
executor = ThreadPoolExecutor(max_workers=1)
queue: asyncio.Queue = asyncio.Queue() # type: ignore queue: asyncio.Queue = asyncio.Queue() # type: ignore
@ -30,7 +33,8 @@ async def bot_startup() -> None:
logger.info(f'Webhook set to {WEBHOOK_URL}'.replace(API_TOKEN, '{BOT_API_TOKEN}')) logger.info(f'Webhook set to {WEBHOOK_URL}'.replace(API_TOKEN, '{BOT_API_TOKEN}'))
asyncio_schedule() asyncio_schedule()
logger.info('Scheduler started') logger.info('Scheduler started')
await worker() loop = asyncio.get_running_loop()
await loop.run_in_executor(executor, worker, queue) # type: ignore
logger.info('Worker started') logger.info('Worker started')
@ -94,14 +98,14 @@ async def webhook(request: web.Request) -> web.Response:
return web.Response(status=HTTPStatus.ACCEPTED) return web.Response(status=HTTPStatus.ACCEPTED)
async def worker() -> None: async def worker(shared_queue: asyncio.Queue) -> None: # type: ignore
Dispatcher.set_current(dispatcher) Dispatcher.set_current(dispatcher)
Bot.set_current(dispatcher.bot) Bot.set_current(dispatcher.bot)
await asyncio.sleep(3) await asyncio.sleep(3)
logger.info('Worker is working') logger.info('Worker is working')
while True: while True:
await asyncio.sleep(4) await asyncio.sleep(4)
update = await queue.get() update = await shared_queue.get()
logger.warning(f"Get update {update}") logger.warning(f"Get update {update}")
await dispatcher.process_update(update) await dispatcher.process_update(update)