Implement config and routing for multiple event writers

1 files changed, 11 insertions, 4 deletions

diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 8391728bdb..54e3889a00 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -918,7 +918,8 @@ class FederationHandler(BaseHandler):
                 )
             )
 
-        await self._handle_new_events(dest, ev_infos, backfilled=True)
+        if ev_infos:
+            await self._handle_new_events(dest, ev_infos, backfilled=True)
 
         # Step 2: Persist the rest of the events in the chunk one by one
         events.sort(key=lambda e: e.depth)
@@ -1366,7 +1367,9 @@ class FederationHandler(BaseHandler):
             #
             # TODO: Currently the events stream is written to from master
             await self._replication.wait_for_stream_position(
-                self.config.worker.writers.events, "events", max_stream_id
+                self.config.worker.events_shard_config.get_instance(room_id),
+                "events",
+                max_stream_id,
             )
 
             predecessor = None
@@ -2916,9 +2919,13 @@ class FederationHandler(BaseHandler):
             backfilled: Whether these events are a result of
                 backfilling or not
         """
-        if self.config.worker.writers.events != self._instance_name:
+        # FIXME:
+        instance = self.config.worker.events_shard_config.get_instance(
+            event_and_contexts[0][0].room_id
+        )
+        if instance != self._instance_name:
             result = await self._send_events(
-                instance_name=self.config.worker.writers.events,
+                instance_name=instance,
                 store=self.store,
                 event_and_contexts=event_and_contexts,
                 backfilled=backfilled,