diff --git a/synapse/app/federation_reader.py b/synapse/app/federation_reader.py
index 90a4816753..eb392e1c9d 100644
--- a/synapse/app/federation_reader.py
+++ b/synapse/app/federation_reader.py
@@ -27,11 +27,11 @@ from synapse.replication.slave.storage.keys import SlavedKeyStore
from synapse.replication.slave.storage.room import RoomStore
from synapse.replication.slave.storage.transactions import TransactionStore
from synapse.replication.slave.storage.directory import DirectoryStore
+from synapse.replication.tcp.client import ReplicationClientHandler
from synapse.server import HomeServer
from synapse.storage.engines import create_engine
-from synapse.util.async import sleep
from synapse.util.httpresourcetree import create_resource_tree
-from synapse.util.logcontext import LoggingContext
+from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
from synapse.util.manhole import manhole
from synapse.util.rlimit import change_resource_limit
from synapse.util.versionstring import get_version_string
@@ -42,7 +42,7 @@ from synapse.crypto import context_factory
from synapse import events
-from twisted.internet import reactor, defer
+from twisted.internet import reactor
from twisted.web.resource import Resource
from daemonize import Daemonize
@@ -134,21 +134,10 @@ class FederationReaderServer(HomeServer):
else:
logger.warn("Unrecognized listener type: %s", listener["type"])
- @defer.inlineCallbacks
- def replicate(self):
- http_client = self.get_simple_http_client()
- store = self.get_datastore()
- replication_url = self.config.worker_replication_url
+ self.get_tcp_replication().start_replication(self)
- while True:
- try:
- args = store.stream_positions()
- args["timeout"] = 30000
- result = yield http_client.get_json(replication_url, args=args)
- yield store.process_replication(result)
- except:
- logger.exception("Error replicating from %r", replication_url)
- yield sleep(5)
+ def build_tcp_replication(self):
+ return ReplicationClientHandler(self.get_datastore())
def start(config_options):
@@ -162,7 +151,7 @@ def start(config_options):
assert config.worker_app == "synapse.app.federation_reader"
- setup_logging(config.worker_log_config, config.worker_log_file)
+ setup_logging(config, use_worker_options=True)
events.USE_FROZEN_DICTS = config.use_frozen_dicts
@@ -184,7 +173,11 @@ def start(config_options):
ss.start_listening(config.worker_listeners)
def run():
- with LoggingContext("run"):
+ # make sure that we run the reactor with the sentinel log context,
+ # otherwise other PreserveLoggingContext instances will get confused
+ # and complain when they see the logcontext arbitrarily swapping
+ # between the sentinel and `run` logcontexts.
+ with PreserveLoggingContext():
logger.info("Running")
change_resource_limit(config.soft_file_limit)
if config.gc_thresholds:
@@ -194,7 +187,6 @@ def start(config_options):
def start():
ss.get_state_handler().start_caching()
ss.get_datastore().start_profiling()
- ss.replicate()
reactor.callWhenRunning(start)
|
