summary refs log tree commit diff
path: root/synapse/replication/tcp/client.py
diff options
context:
space:
mode:
authorBen Banfield-Zanin <benbz@matrix.org>2020-10-15 14:48:13 +0100
committerBen Banfield-Zanin <benbz@matrix.org>2020-10-15 14:48:13 +0100
commit8d9ae573f33110e0420204bceb111fd8df649e7c (patch)
treec8113c67df9769a14e8bb0a03620026dbe9aa0ba /synapse/replication/tcp/client.py
parentMerge remote-tracking branch 'origin/anoa/3pid_check_invite_exemption' into b... (diff)
parentRemove racey assertion in MultiWriterIDGenerator (#8530) (diff)
downloadsynapse-github/bbz/info-mainline-1.21.2.tar.xz
Merge remote-tracking branch 'origin/release-v1.21.2' into bbz/info-mainline-1.21.2 github/bbz/info-mainline-1.21.2 bbz/info-mainline-1.21.2
Diffstat (limited to 'synapse/replication/tcp/client.py')
-rw-r--r--synapse/replication/tcp/client.py14
1 files changed, 8 insertions, 6 deletions
diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py

index d6ecf5b327..e165429cad 100644 --- a/synapse/replication/tcp/client.py +++ b/synapse/replication/tcp/client.py
@@ -29,6 +29,7 @@ from synapse.replication.tcp.streams.events import ( EventsStreamEventRow, EventsStreamRow, ) +from synapse.types import PersistedEventPosition, UserID from synapse.util.async_helpers import timeout_deferred from synapse.util.metrics import Measure @@ -98,7 +99,6 @@ class ReplicationDataHandler: def __init__(self, hs: "HomeServer"): self.store = hs.get_datastore() - self.pusher_pool = hs.get_pusherpool() self.notifier = hs.get_notifier() self._reactor = hs.get_reactor() self._clock = hs.get_clock() @@ -148,13 +148,15 @@ class ReplicationDataHandler: if event.rejected_reason: continue - extra_users = () # type: Tuple[str, ...] + extra_users = () # type: Tuple[UserID, ...] if event.type == EventTypes.Member: - extra_users = (event.state_key,) - max_token = self.store.get_room_max_stream_ordering() - self.notifier.on_new_room_event(event, token, max_token, extra_users) + extra_users = (UserID.from_string(event.state_key),) - await self.pusher_pool.on_new_notifications(token, token) + max_token = self.store.get_room_max_token() + event_pos = PersistedEventPosition(instance_name, token) + self.notifier.on_new_room_event( + event, event_pos, max_token, extra_users + ) # Notify any waiting deferreds. The list is ordered by position so we # just iterate through the list until we reach a position that is