diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py
index ce5d651cb8..4f3c6a18b6 100644
--- a/synapse/replication/tcp/client.py
+++ b/synapse/replication/tcp/client.py
@@ -29,7 +29,6 @@ from synapse.replication.tcp.streams import (
AccountDataStream,
DeviceListsStream,
GroupServerStream,
- PresenceStream,
PushersStream,
PushRulesStream,
ReceiptsStream,
@@ -191,8 +190,6 @@ class ReplicationDataHandler:
self.stop_pusher(row.user_id, row.app_id, row.pushkey)
else:
await self.start_pusher(row.user_id, row.app_id, row.pushkey)
- elif stream_name == PresenceStream.NAME:
- await self._presence_handler.process_replication_rows(token, rows)
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.
@@ -221,6 +218,10 @@ class ReplicationDataHandler:
membership=row.data.membership,
)
+ await self._presence_handler.process_replication_rows(
+ stream_name, instance_name, token, rows
+ )
+
# Notify any waiting deferreds. The list is ordered by position so we
# just iterate through the list until we reach a position that is
# greater than the received row position.
diff --git a/synapse/replication/tcp/streams/__init__.py b/synapse/replication/tcp/streams/__init__.py
index fb74ac4e98..4c0023c68a 100644
--- a/synapse/replication/tcp/streams/__init__.py
+++ b/synapse/replication/tcp/streams/__init__.py
@@ -30,6 +30,7 @@ from synapse.replication.tcp.streams._base import (
CachesStream,
DeviceListsStream,
GroupServerStream,
+ PresenceFederationStream,
PresenceStream,
PublicRoomsStream,
PushersStream,
@@ -50,6 +51,7 @@ STREAMS_MAP = {
EventsStream,
BackfillStream,
PresenceStream,
+ PresenceFederationStream,
TypingStream,
ReceiptsStream,
PushRulesStream,
@@ -71,6 +73,7 @@ __all__ = [
"Stream",
"BackfillStream",
"PresenceStream",
+ "PresenceFederationStream",
"TypingStream",
"ReceiptsStream",
"PushRulesStream",
diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py
index 520c45f151..9d75a89f1c 100644
--- a/synapse/replication/tcp/streams/_base.py
+++ b/synapse/replication/tcp/streams/_base.py
@@ -290,6 +290,30 @@ class PresenceStream(Stream):
)
+class PresenceFederationStream(Stream):
+ """A stream used to send ad hoc presence updates over federation.
+
+ Streams the remote destination and the user ID of the presence state to
+ send.
+ """
+
+ @attr.s(slots=True, auto_attribs=True)
+ class PresenceFederationStreamRow:
+ destination: str
+ user_id: str
+
+ NAME = "presence_federation"
+ ROW_TYPE = PresenceFederationStreamRow
+
+ def __init__(self, hs: "HomeServer"):
+ federation_queue = hs.get_presence_handler().get_federation_queue()
+ super().__init__(
+ hs.get_instance_name(),
+ federation_queue.get_current_token,
+ federation_queue.get_replication_rows,
+ )
+
+
class TypingStream(Stream):
TypingStreamRow = namedtuple(
"TypingStreamRow", ("room_id", "user_ids") # str # list(str)
|
