diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/federation/replication.py | 50 | ||||
| -rw-r--r-- | synapse/federation/transport.py | 25 |
2 files changed, 75 insertions, 0 deletions
diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index bea5335f89..01020566cf 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -66,6 +66,7 @@ class ReplicationLayer(object): self.handler = None self.edu_handlers = {} + self.query_handlers = {} self._order = 0 @@ -84,6 +85,27 @@ class ReplicationLayer(object): self.edu_handlers[edu_type] = handler + def register_query_handler(self, query_type, handler): + """Sets the handler callable that will be used to handle an incoming + federation Query of the given type. + + Args: + query_type (str): Category name of the query, which should match + the string used by make_query. + handler (callable): Invoked to handle incoming queries of this type + + handler is invoked as: + result = handler(args) + + where 'args' is a dict mapping strings to strings of the query + arguments. It should return a Deferred that will eventually yield an + object to encode as JSON. + """ + if query_type in self.query_handlers: + raise KeyError("Already have a Query handler for %s" % (query_type)) + + self.query_handlers[query_type] = handler + @defer.inlineCallbacks @log_function def send_pdu(self, pdu): @@ -137,6 +159,24 @@ class ReplicationLayer(object): # TODO, add errback, etc. self._transaction_queue.enqueue_edu(edu) + @log_function + def make_query(self, destination, query_type, args): + """Sends a federation Query to a remote homeserver of the given type + and arguments. + + Args: + destination (str): Domain name of the remote homeserver + query_type (str): Category of the query type; should match the + handler name used in register_query_handler(). + args (dict): Mapping of strings to strings containing the details + of the query request. + + Returns: + a Deferred which will eventually yield a JSON object from the + response + """ + return self.transport_layer.make_query(destination, query_type, args) + @defer.inlineCallbacks @log_function def paginate(self, dest, context, limit): @@ -341,6 +381,16 @@ class ReplicationLayer(object): ) @defer.inlineCallbacks + def on_query_request(self, query_type, args): + if query_type in self.query_handlers: + response = yield self.query_handlers[query_type](args) + defer.returnValue((200, response)) + else: + defer.returnValue((404, "No handler for Query type '%s'" + % (query_type) + )) + + @defer.inlineCallbacks @log_function def _get_persisted_pdu(self, pdu_id, pdu_origin): """ Get a PDU from the database with given origin and id. diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index ff3fc34419..69166036fb 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -180,6 +180,19 @@ class TransportLayer(object): defer.returnValue((code, response)) + @defer.inlineCallbacks + @log_function + def make_query(self, destination, query_type, args): + path = PREFIX + "/query/%s" % query_type + + response = yield self.client.get_json( + destination=destination, + path=path, + args=args + ) + + defer.returnValue(response) + @log_function def register_received_handler(self, handler): """ Register a handler that will be fired when we receive data. @@ -251,6 +264,15 @@ class TransportLayer(object): lambda request, context: handler.on_context_pdus_request(context) ) + # This is when we receive a server-server Query + self.server.register_path( + "GET", + re.compile("^" + PREFIX + "/query/([^/]*)$"), + lambda request, query_type: handler.on_query_request( + query_type, {k: v[0] for k, v in request.args.items()} + ) + ) + @defer.inlineCallbacks @log_function def _on_send_request(self, request, transaction_id): @@ -456,3 +478,6 @@ class TransportRequestHandler(object): of what went wrong. """ pass + + def on_query_request(self): + """ Called on a GET /query/<query_type> request. """ |