Rename get_tcp_replication to get_replication_command_handler. (#12192)

Since the object it returns is a ReplicationCommandHandler.

This is clean-up from adding support to Redis where the command handler
was added as an additional layer of abstraction from the TCP protocol.
This commit is contained in:
Patrick Cloke 2022-03-10 08:01:56 -05:00 committed by GitHub
parent a4c1fdb44a
commit 3e4af36bc8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 20 additions and 19 deletions

1
changelog.d/12192.misc Normal file
View File

@ -0,0 +1 @@
Rename `HomeServer.get_tcp_replication` to `get_replication_command_handler`.

View File

@ -417,7 +417,7 @@ class GenericWorkerServer(HomeServer):
else: else:
logger.warning("Unsupported listener type: %s", listener.type) logger.warning("Unsupported listener type: %s", listener.type)
self.get_tcp_replication().start_replication(self) self.get_replication_command_handler().start_replication(self)
def start(config_options: List[str]) -> None: def start(config_options: List[str]) -> None:

View File

@ -273,7 +273,7 @@ class SynapseHomeServer(HomeServer):
# If redis is enabled we connect via the replication command handler # If redis is enabled we connect via the replication command handler
# in the same way as the workers (since we're effectively a client # in the same way as the workers (since we're effectively a client
# rather than a server). # rather than a server).
self.get_tcp_replication().start_replication(self) self.get_replication_command_handler().start_replication(self)
for listener in self.config.server.listeners: for listener in self.config.server.listeners:
if listener.type == "http": if listener.type == "http":

View File

@ -63,7 +63,7 @@ class Authenticator:
self.replication_client = None self.replication_client = None
if hs.config.worker.worker_app: if hs.config.worker.worker_app:
self.replication_client = hs.get_tcp_replication() self.replication_client = hs.get_replication_command_handler()
# A method just so we can pass 'self' as the authenticator to the Servlets # A method just so we can pass 'self' as the authenticator to the Servlets
async def authenticate_request( async def authenticate_request(

View File

@ -424,13 +424,13 @@ class WorkerPresenceHandler(BasePresenceHandler):
async def _on_shutdown(self) -> None: async def _on_shutdown(self) -> None:
if self._presence_enabled: if self._presence_enabled:
self.hs.get_tcp_replication().send_command( self.hs.get_replication_command_handler().send_command(
ClearUserSyncsCommand(self.instance_id) ClearUserSyncsCommand(self.instance_id)
) )
def send_user_sync(self, user_id: str, is_syncing: bool, last_sync_ms: int) -> None: def send_user_sync(self, user_id: str, is_syncing: bool, last_sync_ms: int) -> None:
if self._presence_enabled: if self._presence_enabled:
self.hs.get_tcp_replication().send_user_sync( self.hs.get_replication_command_handler().send_user_sync(
self.instance_id, user_id, is_syncing, last_sync_ms self.instance_id, user_id, is_syncing, last_sync_ms
) )

View File

@ -54,6 +54,6 @@ class SlavedClientIpStore(BaseSlavedStore):
self.client_ip_last_seen.set(key, now) self.client_ip_last_seen.set(key, now)
self.hs.get_tcp_replication().send_user_ip( self.hs.get_replication_command_handler().send_user_ip(
user_id, access_token, ip, user_agent, device_id, now user_id, access_token, ip, user_agent, device_id, now
) )

View File

@ -462,6 +462,8 @@ class FederationSenderHandler:
# We ACK this token over replication so that the master can drop # We ACK this token over replication so that the master can drop
# its in memory queues # its in memory queues
self._hs.get_tcp_replication().send_federation_ack(current_position) self._hs.get_replication_command_handler().send_federation_ack(
current_position
)
except Exception: except Exception:
logger.exception("Error updating federation stream position") logger.exception("Error updating federation stream position")

View File

@ -295,9 +295,7 @@ class ReplicationCommandHandler:
raise Exception("Unrecognised command %s in stream queue", cmd.NAME) raise Exception("Unrecognised command %s in stream queue", cmd.NAME)
def start_replication(self, hs: "HomeServer") -> None: def start_replication(self, hs: "HomeServer") -> None:
"""Helper method to start a replication connection to the remote server """Helper method to start replication."""
using TCP.
"""
if hs.config.redis.redis_enabled: if hs.config.redis.redis_enabled:
from synapse.replication.tcp.redis import ( from synapse.replication.tcp.redis import (
RedisDirectTcpReplicationClientFactory, RedisDirectTcpReplicationClientFactory,

View File

@ -325,7 +325,7 @@ class RedisDirectTcpReplicationClientFactory(SynapseRedisFactory):
password=hs.config.redis.redis_password, password=hs.config.redis.redis_password,
) )
self.synapse_handler = hs.get_tcp_replication() self.synapse_handler = hs.get_replication_command_handler()
self.synapse_stream_name = hs.hostname self.synapse_stream_name = hs.hostname
self.synapse_outbound_redis_connection = outbound_redis_connection self.synapse_outbound_redis_connection = outbound_redis_connection

View File

@ -44,7 +44,7 @@ class ReplicationStreamProtocolFactory(ServerFactory):
"""Factory for new replication connections.""" """Factory for new replication connections."""
def __init__(self, hs: "HomeServer"): def __init__(self, hs: "HomeServer"):
self.command_handler = hs.get_tcp_replication() self.command_handler = hs.get_replication_command_handler()
self.clock = hs.get_clock() self.clock = hs.get_clock()
self.server_name = hs.config.server.server_name self.server_name = hs.config.server.server_name
@ -85,7 +85,7 @@ class ReplicationStreamer:
self.is_looping = False self.is_looping = False
self.pending_updates = False self.pending_updates = False
self.command_handler = hs.get_tcp_replication() self.command_handler = hs.get_replication_command_handler()
# Set of streams to replicate. # Set of streams to replicate.
self.streams = self.command_handler.get_streams_to_replicate() self.streams = self.command_handler.get_streams_to_replicate()

View File

@ -639,7 +639,7 @@ class HomeServer(metaclass=abc.ABCMeta):
return ReadMarkerHandler(self) return ReadMarkerHandler(self)
@cache_in_self @cache_in_self
def get_tcp_replication(self) -> ReplicationCommandHandler: def get_replication_command_handler(self) -> ReplicationCommandHandler:
return ReplicationCommandHandler(self) return ReplicationCommandHandler(self)
@cache_in_self @cache_in_self

View File

@ -251,7 +251,7 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
self.connect_any_redis_attempts, self.connect_any_redis_attempts,
) )
self.hs.get_tcp_replication().start_replication(self.hs) self.hs.get_replication_command_handler().start_replication(self.hs)
# When we see a connection attempt to the master replication listener we # When we see a connection attempt to the master replication listener we
# automatically set up the connection. This is so that tests don't # automatically set up the connection. This is so that tests don't
@ -375,7 +375,7 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
) )
if worker_hs.config.redis.redis_enabled: if worker_hs.config.redis.redis_enabled:
worker_hs.get_tcp_replication().start_replication(worker_hs) worker_hs.get_replication_command_handler().start_replication(worker_hs)
return worker_hs return worker_hs

View File

@ -420,7 +420,7 @@ class EventsStreamTestCase(BaseStreamTestCase):
# Manually send an old RDATA command, which should get dropped. This # Manually send an old RDATA command, which should get dropped. This
# re-uses the row from above, but with an earlier stream token. # re-uses the row from above, but with an earlier stream token.
self.hs.get_tcp_replication().send_command( self.hs.get_replication_command_handler().send_command(
RdataCommand("events", "master", 1, row) RdataCommand("events", "master", 1, row)
) )

View File

@ -118,7 +118,7 @@ class TypingStreamTestCase(BaseStreamTestCase):
# Reset the typing handler # Reset the typing handler
self.hs.get_replication_streams()["typing"].last_token = 0 self.hs.get_replication_streams()["typing"].last_token = 0
self.hs.get_tcp_replication()._streams["typing"].last_token = 0 self.hs.get_replication_command_handler()._streams["typing"].last_token = 0
typing._latest_room_serial = 0 typing._latest_room_serial = 0
typing._typing_stream_change_cache = StreamChangeCache( typing._typing_stream_change_cache = StreamChangeCache(
"TypingStreamChangeCache", typing._latest_room_serial "TypingStreamChangeCache", typing._latest_room_serial

View File

@ -48,7 +48,7 @@ class FederationAckTestCase(HomeserverTestCase):
transport, rather than assuming that the implementation has a transport, rather than assuming that the implementation has a
ReplicationCommandHandler. ReplicationCommandHandler.
""" """
rch = self.hs.get_tcp_replication() rch = self.hs.get_replication_command_handler()
# wire up the ReplicationCommandHandler to a mock connection, which needs # wire up the ReplicationCommandHandler to a mock connection, which needs
# to implement IReplicationConnection. (Note that Mock doesn't understand # to implement IReplicationConnection. (Note that Mock doesn't understand