1 files changed, 47 insertions, 2 deletions
diff --git a/synapse/config/workers.py b/synapse/config/workers.py
index bc0fc165e3..ed06b91a54 100644
--- a/synapse/config/workers.py
+++ b/synapse/config/workers.py
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
-# Copyright 2016 matrix.org
+# Copyright 2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -13,7 +13,29 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from ._base import Config
+import attr
+
+from ._base import Config, ConfigError
+
+
+@attr.s
+class InstanceLocationConfig:
+ """The host and port to talk to an instance via HTTP replication.
+ """
+
+ host = attr.ib(type=str)
+ port = attr.ib(type=int)
+
+
+@attr.s
+class WriterLocations:
+ """Specifies the instances that write various streams.
+
+ Attributes:
+ events: The instance that writes to the event and backfill streams.
+ """
+
+ events = attr.ib(default="master", type=str)
class WorkerConfig(Config):
@@ -21,6 +43,8 @@ class WorkerConfig(Config):
They have their own pid_file and listener configuration. They use the
replication_url to talk to the main synapse process."""
+ section = "worker"
+
def read_config(self, config, **kwargs):
self.worker_app = config.get("worker_app")
@@ -69,6 +93,27 @@ class WorkerConfig(Config):
elif not bind_addresses:
bind_addresses.append("")
+ # A map from instance name to host/port of their HTTP replication endpoint.
+ instance_map = config.get("instance_map") or {}
+ self.instance_map = {
+ name: InstanceLocationConfig(**c) for name, c in instance_map.items()
+ }
+
+ # Map from type of streams to source, c.f. WriterLocations.
+ writers = config.get("stream_writers") or {}
+ self.writers = WriterLocations(**writers)
+
+ # Check that the configured writer for events also appears in
+ # `instance_map`.
+ if (
+ self.writers.events != "master"
+ and self.writers.events not in self.instance_map
+ ):
+ raise ConfigError(
+ "Instance %r is configured to write events but does not appear in `instance_map` config."
+ % (self.writers.events,)
+ )
+
def read_arguments(self, args):
# We support a bunch of command line arguments that override options in
# the config. A lot of these options have a worker_* prefix when running
|
