diff --git a/synapse/replication/tcp/streams/__init__.py b/synapse/replication/tcp/streams/__init__.py
index d1a61c3314..4c0023c68a 100644
--- a/synapse/replication/tcp/streams/__init__.py
+++ b/synapse/replication/tcp/streams/__init__.py
@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
# Copyright 2019 New Vector Ltd
#
@@ -31,6 +30,7 @@ from synapse.replication.tcp.streams._base import (
CachesStream,
DeviceListsStream,
GroupServerStream,
+ PresenceFederationStream,
PresenceStream,
PublicRoomsStream,
PushersStream,
@@ -51,6 +51,7 @@ STREAMS_MAP = {
EventsStream,
BackfillStream,
PresenceStream,
+ PresenceFederationStream,
TypingStream,
ReceiptsStream,
PushRulesStream,
@@ -72,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 3dfee76743..b03824925a 100644
--- a/synapse/replication/tcp/streams/_base.py
+++ b/synapse/replication/tcp/streams/_base.py
@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
# Copyright 2019 New Vector Ltd
#
@@ -273,15 +272,22 @@ class PresenceStream(Stream):
NAME = "presence"
ROW_TYPE = PresenceStreamRow
- def __init__(self, hs):
+ def __init__(self, hs: "HomeServer"):
store = hs.get_datastore()
- if hs.config.worker_app is None:
- # on the master, query the presence handler
+ if hs.get_instance_name() in hs.config.worker.writers.presence:
+ # on the presence writer, query the presence handler
presence_handler = hs.get_presence_handler()
- update_function = presence_handler.get_all_presence_updates
+
+ from synapse.handlers.presence import PresenceHandler
+
+ assert isinstance(presence_handler, PresenceHandler)
+
+ update_function = (
+ presence_handler.get_all_presence_updates
+ ) # type: UpdateFunction
else:
- # Query master process
+ # Query presence writer process
update_function = make_http_update_function(hs, self.NAME)
super().__init__(
@@ -291,6 +297,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)
diff --git a/synapse/replication/tcp/streams/events.py b/synapse/replication/tcp/streams/events.py
index fa5e37ba7b..e7e87bac92 100644
--- a/synapse/replication/tcp/streams/events.py
+++ b/synapse/replication/tcp/streams/events.py
@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
# Copyright 2019 New Vector Ltd
#
diff --git a/synapse/replication/tcp/streams/federation.py b/synapse/replication/tcp/streams/federation.py
index 9bb8e9e177..096a85d363 100644
--- a/synapse/replication/tcp/streams/federation.py
+++ b/synapse/replication/tcp/streams/federation.py
@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
# Copyright 2019 New Vector Ltd
#
|
