Move connecting logic into ClientReplicationStreamProtocol
1 files changed, 0 insertions, 18 deletions
diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py
index 51f90655d0..e558f90e1a 100644
--- a/synapse/replication/tcp/client.py
+++ b/synapse/replication/tcp/client.py
@@ -89,11 +89,6 @@ class ReplicationClientHandler(object):
# Used for tests.
self.awaiting_syncs = {}
- # Set of stream names that have been subscribe to, but haven't yet
- # caught up with. This is used to track when the client has been fully
- # connected to the remote.
- self.streams_connecting = None
-
# The factory used to create connections.
self.factory = None
@@ -122,12 +117,6 @@ class ReplicationClientHandler(object):
Can be overriden in subclasses to handle more.
"""
- # When we get a `POSITION` command it means we've finished getting
- # missing updates for the given stream, and are now up to date.
- self.streams_connecting.discard(stream_name)
- if not self.streams_connecting:
- self.finished_connecting()
-
return self.store.process_replication_rows(stream_name, token, [])
def on_sync(self, data):
@@ -154,9 +143,6 @@ class ReplicationClientHandler(object):
elif room_account_data:
args["account_data"] = room_account_data
- # Record which streams we're in the process of subscribing to
- self.streams_connecting = set(args.keys())
-
return args
def get_currently_syncing_users(self):
@@ -222,10 +208,6 @@ class ReplicationClientHandler(object):
connection.send_command(cmd)
self.pending_commands = []
- # This will happen if we don't actually subscribe to any streams
- if not self.streams_connecting:
- self.finished_connecting()
-
def finished_connecting(self):
"""Called when we have successfully subscribed and caught up to all
streams we're interested in.
|