USE_ASYNC now observes all indices

toluaina · Aug 12, 2024 · d8855a5 · d8855a5
1 parent bb64021
commit d8855a5
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 7 deletions.
diff --git a/pgsync/sync.py b/pgsync/sync.py
@@ -109,6 +109,7 @@ def __init__(
             self._plugins: Plugins = Plugins("plugins", self.plugins)
         self.query_builder: QueryBuilder = QueryBuilder(verbose=verbose)
         self.count: dict = dict(xlog=0, db=0, redis=0)
+        self.tasks: t.List[asyncio.Task] = []
 
     def validate(self, repl_slots: bool = True) -> None:
         """Perform all validation right away."""
@@ -1295,13 +1296,11 @@ def receive(self) -> None:
             cursor.execute(f'LISTEN "{self.database}"')
             event_loop = asyncio.get_event_loop()
             event_loop.add_reader(self.conn, self.async_poll_db)
-            tasks: list = [
+            self.tasks: t.List[asyncio.Task] = [
                 event_loop.create_task(self.async_poll_redis()),
                 event_loop.create_task(self.async_truncate_slots()),
                 event_loop.create_task(self.async_status()),
             ]
-            event_loop.run_until_complete(asyncio.wait(tasks))
-            event_loop.close()
 
         else:
             # sync up to and produce items in the Redis cache
@@ -1477,6 +1476,7 @@ def main(
                 time.sleep(settings.POLL_INTERVAL)
 
         else:
+            tasks: t.List[asyncio.Task] = []
             for doc in config_loader(config):
                 sync: Sync = Sync(
                     doc,
@@ -1489,6 +1489,12 @@ def main(
                 sync.pull()
                 if daemon:
                     sync.receive()
+                    tasks.extend(sync.tasks)
+
+            if settings.USE_ASYNC:
+                event_loop = asyncio.get_event_loop()
+                event_loop.run_until_complete(asyncio.gather(*tasks))
+                event_loop.close()
 
 
 if __name__ == "__main__":

diff --git a/requirements/base.txt b/requirements/base.txt
@@ -21,13 +21,13 @@ charset-normalizer==3.3.2
     # via requests
 click==8.1.7
     # via -r requirements/base.in
-elastic-transport==8.13.1
+elastic-transport==8.15.0
     # via elasticsearch
 elasticsearch==8.14.0
     # via
     #   -r requirements/base.in
     #   elasticsearch-dsl
-elasticsearch-dsl==8.14.0
+elasticsearch-dsl==8.15.0
     # via -r requirements/base.in
 environs==11.0.0
     # via -r requirements/base.in

diff --git a/requirements/dev.txt b/requirements/dev.txt
@@ -33,13 +33,13 @@ coverage[toml]==7.6.1
     #   pytest-cov
 distlib==0.3.8
     # via virtualenv
-elastic-transport==8.13.1
+elastic-transport==8.15.0
     # via elasticsearch
 elasticsearch==8.14.0
     # via
     #   -r requirements/base.in
     #   elasticsearch-dsl
-elasticsearch-dsl==8.14.0
+elasticsearch-dsl==8.15.0
     # via -r requirements/base.in
 environs==11.0.0
     # via -r requirements/base.in