Implemented event callbacks

main.py

@@ -9,12 +9,12 @@ from pathlib import Path
 
 import config
 import util
-#import bot
-#import logic
 
 from new_bot import MatrixBot
 from bot_types import MatrixBotConfig
 
+import nio.events
+
 from datatypes import AppConfig
 
 async def main() -> None:
@@ -35,16 +35,22 @@ async def main() -> None:
         util.log_error("Could't load config")
         return
 
+    # setup the bot
     matrix_bot_config = MatrixBotConfig(
         matrix_homeserver_url=cfg.matrix_homeserver,
         matrix_username_localpart=cfg.matrix_user,
         storage_directory=Path(cfg.store_dir)
     )
     bot = MatrixBot(matrix_bot_config)
+    bot.add_event_callback(None, nio.events.Event)
+
+    # start the app
     bot.start()
 
+    # wait for Ctrl+C
     await util.get_app_stop_event().wait()
 
+    # stop the app
     await bot.stop()
 
 if __name__ == "__main__":

new_bot.py

@@ -8,11 +8,13 @@ import json
 import os
 import re
 from urllib.parse import urlparse
-from typing import Any
+from typing import Any, Callable, Awaitable
 
 from nio import AsyncClient, AsyncClientConfig, SyncResponse
 from nio import LoginResponse, LoginError, WhoamiResponse, WhoamiError
 
+import nio.events
+
 from bot_types import *
 
 
@@ -23,6 +25,7 @@ class MatrixBot:
         by yourself.
     """
     NEXT_BATCH_DUMP_PERIOD = 120.0
+    MATRIX_SYNC_PERIOD = 5000
 
     #
     # PRIVATE
@@ -127,6 +130,18 @@ class MatrixBot:
             return os.environ["MATRIX_PASSWORD"]
         print("--- A password is required (btw you might use MATRIX_PASSWORD envvar) ---")
         return await aioconsole.ainput("Password: ")
+
+    @staticmethod
+    async def _debug_event_callback(*args, **kwargs) -> None:
+        """Just prints types of arguments"""
+        try:
+            print(f"_debug_event_callback ({len(args)} args, {len(kwargs)} kwargs)")
+            for a in args:
+                print(f"  - {type(a)}")
+            for k in kwargs:
+                print(f"  * {k} = {kwargs[k]}")
+        except:
+            traceback.print_exc()
     
 
     async def _write_next_batch(self, next_batch: str) -> None:
@@ -263,13 +278,28 @@ class MatrixBot:
         except:
             traceback.print_exc()
 
+    async def _client_cancellable_sync_forever(self, *args, **kwargs) -> Any:
+        """Behaves exactly like AsyncClient.sync_forever, but supports task cancellation"""
+        sync_forever_task = asyncio.create_task(
+            self._client.sync_forever(*args, **kwargs)
+        )
+        try:
+            return await sync_forever_task
+        except asyncio.CancelledError:
+            try:
+                self._client.stop_sync_forever()
+            except Exception:
+                pass
+            finally:
+                sync_forever_task.cancel()
+                await asyncio.gather(sync_forever_task, return_exceptions=True)
+            raise
+
     async def _background_coroutine(self) -> None:
         """This function implements bot lifecycle."""
         # we should stop when this task stops
         self._logger.debug("_background_coroutine is started")
         stop_wait_task = asyncio.create_task(self._stop_event.wait())
-        self._client = self._build_client(self._config)
-        self._setup_client_callbacks()
         # perform login
         login_task = asyncio.create_task(self._client_login())
         try:
@@ -281,13 +311,18 @@ class MatrixBot:
         except:
             traceback.print_exc()
         self._logger.info("Succesfully logged in")
-        # test
+        # sync forever
-        wait_task = asyncio.create_task(asyncio.sleep(100000))
+        self._logger.info("Syncing forever")
+        sync_task = asyncio.create_task(
+            self._client_cancellable_sync_forever(
+                timeout=self.MATRIX_SYNC_PERIOD,
+                since=(await self._read_next_batch())
+            )
+        )
         try:
-            self._logger.info("Bot is not implemented yet, sleeping forever")
+            await self._wait_for_task_and_stop_event(sync_task, stop_wait_task)
-            await self._wait_for_task_and_stop_event(wait_task, stop_wait_task)
         except asyncio.CancelledError:
-            self._logger.info("Background task is cancelled during eternal sleep")
+            self._logger.debug("Sync task is cancelled")
             await self._client_destroy()
             return
         except:
@@ -301,14 +336,16 @@ class MatrixBot:
         # check if config is valid
         self._validate_bot_config(config) # may raise an Exception
         # save the config
-        self._config = config
+        self._config: MatrixBotConfig = config
         # create the logger
         self._logger = logging.getLogger(self._build_matrix_username(config))
         self._logger.setLevel(logging.DEBUG)
+        # create the client
+        self._client: AsyncClient = self._build_client(self._config)
+        self._setup_client_callbacks()
 
         # prepare some private data
         self._background_task: asyncio.Task | None = None
-        self._client: AsyncClient = None # type: ignore
         self._last_next_batch_dump: float = 0.0
         self._last_next_batch: str | None = None
         self._cb_password = self._default_password_callback
@@ -337,4 +374,11 @@ class MatrixBot:
         except:
             traceback.print_exc()
         self._stop_event = None
-        self._background_task = None
+        self._background_task = None
+
+    def add_event_callback(self, callback: Callable[[Any], Awaitable[None]] | None, event_class: nio.events.Event) -> None:
+        """Added event callback for events of specified class.
+            Use `None` instead of callback to print parameter types you need to use in your callback."""
+        if callback is None:
+            callback = self._debug_event_callback
+        self._client.add_event_callback(callback, event_class) # type: ignore