Merge branch 'develop' of github.com:matrix-org/synapse into timeout-federation-requests

1 files changed, 221 insertions, 0 deletions

diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py
new file mode 100644
index 0000000000..8b137e7128
--- /dev/null
+++ b/synapse/federation/transport/client.py
@@ -0,0 +1,221 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014, 2015 OpenMarket Ltd
+#
+# 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 twisted.internet import defer
+
+from synapse.api.urls import FEDERATION_PREFIX as PREFIX
+from synapse.util.logutils import log_function
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class TransportLayerClient(object):
+    """Sends federation HTTP requests to other servers"""
+
+    @log_function
+    def get_room_state(self, destination, room_id, event_id):
+        """ Requests all state for a given room from the given server at the
+        given event.
+
+        Args:
+            destination (str): The host name of the remote home server we want
+                to get the state from.
+            context (str): The name of the context we want the state of
+            event_id (str): The event we want the context at.
+
+        Returns:
+            Deferred: Results in a dict received from the remote homeserver.
+        """
+        logger.debug("get_room_state dest=%s, room=%s",
+                     destination, room_id)
+
+        path = PREFIX + "/state/%s/" % room_id
+        return self.client.get_json(
+            destination, path=path, args={"event_id": event_id},
+        )
+
+    @log_function
+    def get_event(self, destination, event_id):
+        """ Requests the pdu with give id and origin from the given server.
+
+        Args:
+            destination (str): The host name of the remote home server we want
+                to get the state from.
+            event_id (str): The id of the event being requested.
+
+        Returns:
+            Deferred: Results in a dict received from the remote homeserver.
+        """
+        logger.debug("get_pdu dest=%s, event_id=%s",
+                     destination, event_id)
+
+        path = PREFIX + "/event/%s/" % (event_id, )
+        return self.client.get_json(destination, path=path)
+
+    @log_function
+    def backfill(self, destination, room_id, event_tuples, limit):
+        """ Requests `limit` previous PDUs in a given context before list of
+        PDUs.
+
+        Args:
+            dest (str)
+            room_id (str)
+            event_tuples (list)
+            limt (int)
+
+        Returns:
+            Deferred: Results in a dict received from the remote homeserver.
+        """
+        logger.debug(
+            "backfill dest=%s, room_id=%s, event_tuples=%s, limit=%s",
+            destination, room_id, repr(event_tuples), str(limit)
+        )
+
+        if not event_tuples:
+            # TODO: raise?
+            return
+
+        path = PREFIX + "/backfill/%s/" % (room_id,)
+
+        args = {
+            "v": event_tuples,
+            "limit": [str(limit)],
+        }
+
+        return self.client.get_json(
+            destination,
+            path=path,
+            args=args,
+        )
+
+    @defer.inlineCallbacks
+    @log_function
+    def send_transaction(self, transaction, json_data_callback=None):
+        """ Sends the given Transaction to its destination
+
+        Args:
+            transaction (Transaction)
+
+        Returns:
+            Deferred: Results of the deferred is a tuple in the form of
+            (response_code, response_body) where the response_body is a
+            python dict decoded from json
+        """
+        logger.debug(
+            "send_data dest=%s, txid=%s",
+            transaction.destination, transaction.transaction_id
+        )
+
+        if transaction.destination == self.server_name:
+            raise RuntimeError("Transport layer cannot send to itself!")
+
+        # FIXME: This is only used by the tests. The actual json sent is
+        # generated by the json_data_callback.
+        json_data = transaction.get_dict()
+
+        response = yield self.client.put_json(
+            transaction.destination,
+            path=PREFIX + "/send/%s/" % transaction.transaction_id,
+            data=json_data,
+            json_data_callback=json_data_callback,
+        )
+
+        logger.debug(
+            "send_data dest=%s, txid=%s, got response: 200",
+            transaction.destination, transaction.transaction_id,
+        )
+
+        defer.returnValue(response)
+
+    @defer.inlineCallbacks
+    @log_function
+    def make_query(self, destination, query_type, args, retry_on_dns_fail):
+        path = PREFIX + "/query/%s" % query_type
+
+        content = yield self.client.get_json(
+            destination=destination,
+            path=path,
+            args=args,
+            retry_on_dns_fail=retry_on_dns_fail,
+        )
+
+        defer.returnValue(content)
+
+    @defer.inlineCallbacks
+    @log_function
+    def make_join(self, destination, room_id, user_id, retry_on_dns_fail=True):
+        path = PREFIX + "/make_join/%s/%s" % (room_id, user_id)
+
+        content = yield self.client.get_json(
+            destination=destination,
+            path=path,
+            retry_on_dns_fail=retry_on_dns_fail,
+        )
+
+        defer.returnValue(content)
+
+    @defer.inlineCallbacks
+    @log_function
+    def send_join(self, destination, room_id, event_id, content):
+        path = PREFIX + "/send_join/%s/%s" % (room_id, event_id)
+
+        response = yield self.client.put_json(
+            destination=destination,
+            path=path,
+            data=content,
+        )
+
+        defer.returnValue(response)
+
+    @defer.inlineCallbacks
+    @log_function
+    def send_invite(self, destination, room_id, event_id, content):
+        path = PREFIX + "/invite/%s/%s" % (room_id, event_id)
+
+        response = yield self.client.put_json(
+            destination=destination,
+            path=path,
+            data=content,
+        )
+
+        defer.returnValue(response)
+
+    @defer.inlineCallbacks
+    @log_function
+    def get_event_auth(self, destination, room_id, event_id):
+        path = PREFIX + "/event_auth/%s/%s" % (room_id, event_id)
+
+        content = yield self.client.get_json(
+            destination=destination,
+            path=path,
+        )
+
+        defer.returnValue(content)
+
+    @defer.inlineCallbacks
+    @log_function
+    def send_query_auth(self, destination, room_id, event_id, content):
+        path = PREFIX + "/query_auth/%s/%s" % (room_id, event_id)
+
+        content = yield self.client.post_json(
+            destination=destination,
+            path=path,
+            data=content,
+        )
+
+        defer.returnValue(content)