3 files changed, 16 insertions, 4 deletions
diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py
index d51f045f22..b217c35f99 100644
--- a/synapse/replication/tcp/handler.py
+++ b/synapse/replication/tcp/handler.py
@@ -709,7 +709,7 @@ class ReplicationCommandHandler:
self.send_command(RemoteServerUpCommand(server))
def stream_update(self, stream_name: str, token: Optional[int], data: Any) -> None:
- """Called when a new update is available to stream to clients.
+ """Called when a new update is available to stream to Redis subscribers.
We need to check if the client is interested in the stream or not
"""
diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py
index ab829040cd..c6870df8f9 100644
--- a/synapse/replication/tcp/resource.py
+++ b/synapse/replication/tcp/resource.py
@@ -67,8 +67,8 @@ class ReplicationStreamProtocolFactory(ServerFactory):
class ReplicationStreamer:
"""Handles replication connections.
- This needs to be poked when new replication data may be available. When new
- data is available it will propagate to all connected clients.
+ This needs to be poked when new replication data may be available.
+ When new data is available it will propagate to all Redis subscribers.
"""
def __init__(self, hs: "HomeServer"):
@@ -109,7 +109,7 @@ class ReplicationStreamer:
def on_notifier_poke(self) -> None:
"""Checks if there is actually any new data and sends it to the
- connections if there are.
+ Redis subscribers if there are.
This should get called each time new data is available, even if it
is currently being executed, so that nothing gets missed
diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py
index 23d631a769..495f2f0285 100644
--- a/synapse/replication/tcp/streams/_base.py
+++ b/synapse/replication/tcp/streams/_base.py
@@ -316,7 +316,19 @@ class PresenceFederationStream(Stream):
class TypingStream(Stream):
@attr.s(slots=True, frozen=True, auto_attribs=True)
class TypingStreamRow:
+ """
+ An entry in the typing stream.
+ Describes all the users that are 'typing' right now in one room.
+
+ When a user stops typing, it will be streamed as a new update with that
+ user absent; you can think of the `user_ids` list as overwriting the
+ entire list that was there previously.
+ """
+
+ # The room that this update is for.
room_id: str
+
+ # All the users that are 'typing' right now in the specified room.
user_ids: List[str]
NAME = "typing"
|
