From a3f5bf79a0fc0ea6d59069945f53717a3e9c6581 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 26 Jul 2018 11:44:22 +0100
Subject: Add EDU/query handling over replication

---
 synapse/federation/federation_server.py | 43 +++++++++++++++++++++++++++++++++
 1 file changed, 43 insertions(+)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index bf89d568af..941e30a596 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -33,6 +33,10 @@ from synapse.federation.federation_base import FederationBase, event_from_pdu_js
 from synapse.federation.persistence import TransactionActions
 from synapse.federation.units import Edu, Transaction
 from synapse.http.endpoint import parse_server_name
+from synapse.replication.http.federation import (
+    ReplicationFederationSendEduRestServlet,
+    ReplicationGetQueryRestServlet,
+)
 from synapse.types import get_domain_from_id
 from synapse.util import async
 from synapse.util.caches.response_cache import ResponseCache
@@ -745,6 +749,8 @@ class FederationHandlerRegistry(object):
         if edu_type in self.edu_handlers:
             raise KeyError("Already have an EDU handler for %s" % (edu_type,))
 
+        logger.info("Registering federation EDU handler for %r", edu_type)
+
         self.edu_handlers[edu_type] = handler
 
     def register_query_handler(self, query_type, handler):
@@ -763,6 +769,8 @@ class FederationHandlerRegistry(object):
                 "Already have a Query handler for %s" % (query_type,)
             )
 
+        logger.info("Registering federation query handler for %r", query_type)
+
         self.query_handlers[query_type] = handler
 
     @defer.inlineCallbacks
@@ -785,3 +793,38 @@ class FederationHandlerRegistry(object):
             raise NotFoundError("No handler for Query type '%s'" % (query_type,))
 
         return handler(args)
+
+
+class ReplicationFederationHandlerRegistry(FederationHandlerRegistry):
+    def __init__(self, hs):
+        self.config = hs.config
+        self.http_client = hs.get_simple_http_client()
+        self.clock = hs.get_clock()
+
+        self._get_query_client = ReplicationGetQueryRestServlet.make_client(hs)
+        self._send_edu = ReplicationFederationSendEduRestServlet.make_client(hs)
+
+        super(ReplicationFederationHandlerRegistry, self).__init__()
+
+    def on_edu(self, edu_type, origin, content):
+        handler = self.edu_handlers.get(edu_type)
+        if handler:
+            return super(ReplicationFederationHandlerRegistry, self).on_edu(
+                edu_type, origin, content,
+            )
+
+        return self._send_edu(
+                edu_type=edu_type,
+                origin=origin,
+                content=content,
+        )
+
+    def on_query(self, query_type, args):
+        handler = self.query_handlers.get(query_type)
+        if handler:
+            return handler(args)
+
+        return self._get_query_client(
+                query_type=query_type,
+                args=args,
+        )
-- 
cgit 1.4.1


From 72d1902bbeaa029e7003aecaeaf9bc5a41d17646 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 9 Aug 2018 10:23:49 +0100
Subject: Fixup doc comments

---
 synapse/federation/federation_server.py | 11 +++++++++++
 synapse/replication/http/federation.py  | 17 +++++++++++++++++
 2 files changed, 28 insertions(+)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index d23c1cf13b..3bc26c4ab4 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -811,6 +811,13 @@ class FederationHandlerRegistry(object):
 
 
 class ReplicationFederationHandlerRegistry(FederationHandlerRegistry):
+    """A FederationHandlerRegistry for worker processes.
+
+    When receiving EDU or queries it will check if an appropriate handler has
+    been registered on the worker, if there isn't one then it calls off to the
+    master process.
+    """
+
     def __init__(self, hs):
         self.config = hs.config
         self.http_client = hs.get_simple_http_client()
@@ -822,6 +829,8 @@ class ReplicationFederationHandlerRegistry(FederationHandlerRegistry):
         super(ReplicationFederationHandlerRegistry, self).__init__()
 
     def on_edu(self, edu_type, origin, content):
+        """Overrides FederationHandlerRegistry
+        """
         handler = self.edu_handlers.get(edu_type)
         if handler:
             return super(ReplicationFederationHandlerRegistry, self).on_edu(
@@ -835,6 +844,8 @@ class ReplicationFederationHandlerRegistry(FederationHandlerRegistry):
         )
 
     def on_query(self, query_type, args):
+        """Overrides FederationHandlerRegistry
+        """
         handler = self.query_handlers.get(query_type)
         if handler:
             return handler(args)
diff --git a/synapse/replication/http/federation.py b/synapse/replication/http/federation.py
index 3fa7bd64c7..3e6cbbf5a1 100644
--- a/synapse/replication/http/federation.py
+++ b/synapse/replication/http/federation.py
@@ -157,6 +157,15 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint):
 class ReplicationFederationSendEduRestServlet(ReplicationEndpoint):
     """Handles EDUs newly received from federation, including persisting and
     notifying.
+
+    Request format:
+
+        POST /_synapse/replication/fed_send_edu/:edu_type/:txn_id
+
+        {
+            "origin": ...,
+            "content: { ... }
+        }
     """
 
     NAME = "fed_send_edu"
@@ -196,6 +205,14 @@ class ReplicationFederationSendEduRestServlet(ReplicationEndpoint):
 
 class ReplicationGetQueryRestServlet(ReplicationEndpoint):
     """Handle responding to queries from federation.
+
+    Request format:
+
+        POST /_synapse/replication/fed_query/:query_type
+
+        {
+            "args": { ... }
+        }
     """
 
     NAME = "fed_query"
-- 
cgit 1.4.1