Fixed some event loop issues.

Signed-off-by: Pavel Kirilin <win10@list.ru>

Author: s3rius

.flake8

@@ -90,6 +90,8 @@ ignore =
   N802,
   ; Do not perform function calls in argument defaults.
   B008,
+  ; Found protected attribute
+  WPS437,
 
   ; all init files
   __init__.py:

taskiq_aiohttp/__init__.py

@@ -1,4 +1,4 @@
 """Project was generated using taskiq."""
 from taskiq_aiohttp.initializer import init
 
-__all__ = ['init']
+__all__ = ["init"]

taskiq_aiohttp/initializer.py

@@ -1,105 +1,131 @@
 import asyncio
 import inspect
 from typing import Awaitable, Callable
-from taskiq import AsyncBroker, TaskiqEvents, TaskiqState
-from taskiq.cli.utils import import_object
+
+import yarl
 from aiohttp import web
-from aiohttp.web_request import RawRequestMessage
-from aiohttp.web_urldispatcher import UrlMappingMatchInfo, ResourceRoute
 from aiohttp.http_writer import HttpVersion10
 from aiohttp.web_protocol import RequestHandler
-import yarl
+from aiohttp.web_request import RawRequestMessage
+from aiohttp.web_urldispatcher import SystemRoute, UrlMappingMatchInfo
+from multidict import CIMultiDict, CIMultiDictProxy
+from taskiq import AsyncBroker, TaskiqEvents, TaskiqState
+from taskiq.cli.utils import import_object
 
-# def _startup(state: Context):
-#     state.broker.add_dependency_context()
 
+def startup_event_generator(
+    broker: AsyncBroker,
+    app_path: str,
+) -> Callable[[TaskiqState], Awaitable[None]]:
+    """
+    Creates an event to run on broker's startup.
 
-async def _mocked_handler(request: web.Request) -> web.Response:
-    return web.Response()
+    This function create a mock application for
+    later use and updates broker's dependency context.
 
+    Also we run application's startup event so it would
+    act the same as the real application.
+
+    :param broker: current broker.
+    :param app_path: string with a path to an application or a factory.
+
+    :returns: a function that is called on startup.
+    """
 
-def startup_event_generator(
-    app: web.Application,
-) -> Callable[[TaskiqState], Awaitable[None]]:
     async def startup(state: TaskiqState) -> None:
+        loop = asyncio.get_event_loop()
+
+        app = import_object(app_path)
+
+        if not isinstance(app, web.Application):
+            app = app()
+
+        if inspect.iscoroutine(app):
+            app = await app
+
+        if not isinstance(app, web.Application):
+            raise ValueError(f"'{app_path}' is not an AioHTTP application.")
+
+        handler = RequestHandler(app._make_handler(), loop=loop)
+        handler.transport = asyncio.Transport()
+        request = web.Request(
+            RawRequestMessage(
+                "GET",
+                "/",
+                HttpVersion10,
+                headers=CIMultiDictProxy(CIMultiDict()),
+                raw_headers=(),
+                should_close=False,
+                upgrade=False,
+                chunked=False,
+                compression=None,
+                url=yarl.URL.build(
+                    scheme="https",
+                    host="test.com",
+                    path="/",
+                ),
+            ),
+            None,
+            handler,
+            None,
+            None,
+            None,
+        )
+
+        request._match_info = UrlMappingMatchInfo(
+            match_dict={},
+            route=SystemRoute(web.HTTPBadRequest()),
+        )
+        request._match_info._apps = app._subapps
+        request._match_info._current_app = app
+
+        broker.add_dependency_context(
+            {
+                web.Application: app,
+                web.Request: request,
+            },
+        )
+
         state.aiohttp_app = app
         app.router._resources = []
         await app.startup()
 
     return startup
 
 
-def shutdown_event_generator(
-    app: web.Application,
-) -> Callable[[TaskiqState], Awaitable[None]]:
-    async def shutdown(_: TaskiqState) -> None:
-        await app.shutdown()
-        await app.cleanup()
+async def shutdown(state: TaskiqState) -> None:
+    """
+    Shuts down the app.
 
-    return shutdown
+    It just gets the application
+    we created in startup and shuts it down.
+
+    :param state: current state.
+    """
+    await state.aiohttp_app.shutdown()
+    await state.aiohttp_app.cleanup()
 
 
 def init(broker: AsyncBroker, app_path: str) -> None:
-    loop = asyncio.get_event_loop()
-    if not broker.is_worker_process:
-        return
+    """
+    Initialize dependencies for your taskiq.
 
-    app = import_object(app_path)
-
-    if not isinstance(app, web.Application):
-        app = app()
-
-    if inspect.isawaitable(app):
-        app = asyncio.run(app)
-
-    if not isinstance(app, web.Application):
-        raise ValueError(f"'{app_path}' is not an AioHTTP application.")
-
-    handler = RequestHandler(app._make_handler(), loop=loop)
-    handler.transport = asyncio.Transport()
-    request = web.Request(
-        RawRequestMessage(
-            "GET",
-            "/",
-            HttpVersion10,
-            headers={},
-            raw_headers=tuple(),
-            should_close=False,
-            upgrade=False,
-            chunked=False,
-            compression=None,
-            url=yarl.URL.build(
-                scheme="https",
-                host="test.com",
-                path="/",
-            ),
-        ),
-        None,
-        handler,
-        None,
-        None,
-        None,
-    )
+    This function imports your application and tries to
+    update broker's dependency context, so request and
+    applicaiton will be available in all your taskiq
+    dependencies.
 
-    request._match_info = UrlMappingMatchInfo(
-        match_dict={},
-        route=ResourceRoute("GET", _mocked_handler, None),
-    )
-    request._match_info._apps = app._subapps
-    request._match_info._current_app = app
-
-    broker.add_dependency_context(
-        {
-            web.Application: app,
-            web.Request: request,
-        }
-    )
+    :param broker: current broker.
+    :param app_path: string with a path to an application or a factory.
+    """
+    if not broker.is_worker_process:
+        return
 
     broker.add_event_handler(
         TaskiqEvents.WORKER_STARTUP,
-        startup_event_generator(app),
+        startup_event_generator(broker, app_path),
     )
     broker.add_event_handler(
         TaskiqEvents.WORKER_SHUTDOWN,
-        shutdown_event_generator(app),
+        shutdown,
     )