summary refs log tree commit diff
path: root/synapse/replication/tcp
diff options
context:
space:
mode:
authorBrendan Abolivier <babolivier@matrix.org>2022-09-21 15:39:01 +0100
committerGitHub <noreply@github.com>2022-09-21 14:39:01 +0000
commit8ae42ab8fa3c6b52d74c24daa7ca75a478fa4fbb (patch)
treedeb3a81241a31a811c26e5bfd64ac56cfc1e83aa /synapse/replication/tcp
parentAdd cache invalidation across workers to module API (#13667) (diff)
downloadsynapse-8ae42ab8fa3c6b52d74c24daa7ca75a478fa4fbb.tar.xz
Support enabling/disabling pushers (from MSC3881) (#13799)
Partial implementation of MSC3881
Diffstat (limited to 'synapse/replication/tcp')
-rw-r--r--synapse/replication/tcp/client.py10
1 files changed, 7 insertions, 3 deletions
diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py
index e4f2201c92..cf9cd6833b 100644
--- a/synapse/replication/tcp/client.py
+++ b/synapse/replication/tcp/client.py
@@ -189,7 +189,9 @@ class ReplicationDataHandler:
                 if row.deleted:
                     self.stop_pusher(row.user_id, row.app_id, row.pushkey)
                 else:
-                    await self.start_pusher(row.user_id, row.app_id, row.pushkey)
+                    await self.process_pusher_change(
+                        row.user_id, row.app_id, row.pushkey
+                    )
         elif stream_name == EventsStream.NAME:
             # We shouldn't get multiple rows per token for events stream, so
             # we don't need to optimise this for multiple rows.
@@ -334,13 +336,15 @@ class ReplicationDataHandler:
         logger.info("Stopping pusher %r / %r", user_id, key)
         pusher.on_stop()
 
-    async def start_pusher(self, user_id: str, app_id: str, pushkey: str) -> None:
+    async def process_pusher_change(
+        self, user_id: str, app_id: str, pushkey: str
+    ) -> None:
         if not self._notify_pushers:
             return
 
         key = "%s:%s" % (app_id, pushkey)
         logger.info("Starting pusher %r / %r", user_id, key)
-        await self._pusher_pool.start_pusher_by_id(app_id, pushkey, user_id)
+        await self._pusher_pool.process_pusher_change_by_id(app_id, pushkey, user_id)
 
 
 class FederationSenderHandler: