diff --git a/synapse/replication/http/_base.py b/synapse/replication/http/_base.py
index 8c2c54c07a..354a907cec 100644
--- a/synapse/replication/http/_base.py
+++ b/synapse/replication/http/_base.py
@@ -198,9 +198,6 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta):
local_instance_name = hs.get_instance_name()
# The value of these option should match the replication listener settings
- master_host = hs.config.worker.worker_replication_host
- master_port = hs.config.worker.worker_replication_http_port
- master_tls = hs.config.worker.worker_replication_http_tls
instance_map = hs.config.worker.instance_map
@@ -221,15 +218,7 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta):
with outgoing_gauge.track_inprogress():
if instance_name == local_instance_name:
raise Exception("Trying to send HTTP request to self")
- if instance_name == "master":
- host = master_host
- port = master_port
- tls = master_tls
- elif instance_name in instance_map:
- host = instance_map[instance_name].host
- port = instance_map[instance_name].port
- tls = instance_map[instance_name].tls
- else:
+ if instance_name not in instance_map:
raise Exception(
"Instance %r not in 'instance_map' config" % (instance_name,)
)
@@ -279,14 +268,10 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta):
# Here the protocol is hard coded to be http by default or https in case the replication
# port is set to have tls true.
+ tls = False # TODO
scheme = "https" if tls else "http"
- uri = "%s://%s:%s/_synapse/replication/%s/%s" % (
- scheme,
- host,
- port,
- cls.NAME,
- "/".join(url_args),
- )
+ joined_args = "/".join(url_args)
+ uri = f"{scheme}://{instance_name}/_synapse/replication/{cls.NAME}/{joined_args}"
headers: Dict[bytes, List[bytes]] = {}
# Add an authorization header, if configured.
diff --git a/synapse/replication/http/endpoint_factory.py b/synapse/replication/http/endpoint_factory.py
new file mode 100644
index 0000000000..87ce468cf5
--- /dev/null
+++ b/synapse/replication/http/endpoint_factory.py
@@ -0,0 +1,67 @@
+# Copyright 2023 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from typing import Dict
+
+from zope.interface import implementer
+
+from twisted.internet.endpoints import UNIXClientEndpoint
+from twisted.internet.interfaces import IStreamClientEndpoint
+from twisted.web.client import URI
+from twisted.web.iweb import IAgentEndpointFactory
+
+from synapse.config.workers import (
+ InstanceLocationConfig,
+ TcpInstanceLocationConfig,
+ UnixSocketInstanceLocationConfig,
+)
+from synapse.types import ISynapseReactor
+
+
+@implementer(IAgentEndpointFactory)
+class WorkerEndpointFactory:
+ def __init__(
+ self,
+ reactor: ISynapseReactor,
+ configs: Dict[str, InstanceLocationConfig],
+ tcp_endpoint_factory: IAgentEndpointFactory,
+ ):
+ self.reactor = reactor
+ self.configs = configs
+ self.tcp_agent_factory = tcp_endpoint_factory
+
+ def endpointForURI(self, uri: URI) -> IStreamClientEndpoint:
+ worker_config = self.configs.get(uri.host)
+ if not worker_config:
+ raise ValueError(f"Don't know how to connect to worker: {uri.host}")
+
+ if isinstance(worker_config, TcpInstanceLocationConfig):
+ # TODO TLS support
+ rewritten_uri = URI(
+ scheme=uri.scheme,
+ # TODO I'd probably cache the encoded netloc and host in the TCP Config?
+ netloc=f"{worker_config.host}:{worker_config.port}".encode("utf-8"),
+ host=worker_config.host.encode("utf-8"),
+ port=worker_config.port,
+ path=uri.path,
+ params=uri.params,
+ query=uri.query,
+ fragment=uri.fragment,
+ )
+ return self.tcp_agent_factory.endpointForURI(rewritten_uri)
+ elif isinstance(worker_config, UnixSocketInstanceLocationConfig):
+ return UNIXClientEndpoint(self.reactor, worker_config.socket_path)
+ else:
+ raise ValueError(
+ f"Unknown worker connection config {worker_config} for {uri.host}"
+ )
|
