diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py
new file mode 100644
index 0000000000..2136adf8d7
--- /dev/null
+++ b/synapse/federation/transport.py
@@ -0,0 +1,454 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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.
+"""The transport layer is responsible for both sending transactions to remote
+home servers and receiving a variety of requests from other home servers.
+
+Typically, this is done over HTTP (and all home servers are required to
+support HTTP), however individual pairings of servers may decide to communicate
+over a different (albeit still reliable) protocol.
+"""
+
+from twisted.internet import defer
+
+from synapse.util.logutils import log_function
+
+import logging
+import json
+import re
+
+
+logger = logging.getLogger(__name__)
+
+
+class TransportLayer(object):
+ """This is a basic implementation of the transport layer that translates
+ transactions and other requests to/from HTTP.
+
+ Attributes:
+ server_name (str): Local home server host
+
+ server (synapse.http.server.HttpServer): the http server to
+ register listeners on
+
+ client (synapse.http.client.HttpClient): the http client used to
+ send requests
+
+ request_handler (TransportRequestHandler): The handler to fire when we
+ receive requests for data.
+
+ received_handler (TransportReceivedHandler): The handler to fire when
+ we receive data.
+ """
+
+ def __init__(self, server_name, server, client):
+ """
+ Args:
+ server_name (str): Local home server host
+ server (synapse.protocol.http.HttpServer): the http server to
+ register listeners on
+ client (synapse.protocol.http.HttpClient): the http client used to
+ send requests
+ """
+ self.server_name = server_name
+ self.server = server
+ self.client = client
+ self.request_handler = None
+ self.received_handler = None
+
+ @log_function
+ def get_context_state(self, destination, context):
+ """ Requests all state for a given context (i.e. room) from the
+ given server.
+
+ 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
+
+ Returns:
+ Deferred: Results in a dict received from the remote homeserver.
+ """
+ logger.debug("get_context_state dest=%s, context=%s",
+ destination, context)
+
+ path = "/state/%s/" % context
+
+ return self._do_request_for_transaction(destination, path)
+
+ @log_function
+ def get_pdu(self, destination, pdu_origin, pdu_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.
+ pdu_origin (str): The home server which created the PDU.
+ pdu_id (str): The id of the PDU being requested.
+
+ Returns:
+ Deferred: Results in a dict received from the remote homeserver.
+ """
+ logger.debug("get_pdu dest=%s, pdu_origin=%s, pdu_id=%s",
+ destination, pdu_origin, pdu_id)
+
+ path = "/pdu/%s/%s/" % (pdu_origin, pdu_id)
+
+ return self._do_request_for_transaction(destination, path)
+
+ @log_function
+ def paginate(self, dest, context, pdu_tuples, limit):
+ """ Requests `limit` previous PDUs in a given context before list of
+ PDUs.
+
+ Args:
+ dest (str)
+ context (str)
+ pdu_tuples (list)
+ limt (int)
+
+ Returns:
+ Deferred: Results in a dict received from the remote homeserver.
+ """
+ logger.debug(
+ "paginate dest=%s, context=%s, pdu_tuples=%s, limit=%s",
+ dest, context, repr(pdu_tuples), str(limit)
+ )
+
+ if not pdu_tuples:
+ return
+
+ path = "/paginate/%s/" % context
+
+ args = {"v": ["%s,%s" % (i, o) for i, o in pdu_tuples]}
+ args["limit"] = limit
+
+ return self._do_request_for_transaction(
+ dest,
+ path,
+ args=args,
+ )
+
+ @defer.inlineCallbacks
+ @log_function
+ def send_transaction(self, transaction):
+ """ Sends the given Transaction to it's 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!")
+
+ data = transaction.get_dict()
+
+ code, response = yield self.client.put_json(
+ transaction.destination,
+ path="/send/%s/" % transaction.transaction_id,
+ data=data
+ )
+
+ logger.debug(
+ "send_data dest=%s, txid=%s, got response: %d",
+ transaction.destination, transaction.transaction_id, code
+ )
+
+ defer.returnValue((code, response))
+
+ @log_function
+ def register_received_handler(self, handler):
+ """ Register a handler that will be fired when we receive data.
+
+ Args:
+ handler (TransportReceivedHandler)
+ """
+ self.received_handler = handler
+
+ # This is when someone is trying to send us a bunch of data.
+ self.server.register_path(
+ "PUT",
+ re.compile("^/send/([^/]*)/$"),
+ self._on_send_request
+ )
+
+ @log_function
+ def register_request_handler(self, handler):
+ """ Register a handler that will be fired when we get asked for data.
+
+ Args:
+ handler (TransportRequestHandler)
+ """
+ self.request_handler = handler
+
+ # TODO(markjh): Namespace the federation URI paths
+
+ # This is for when someone asks us for everything since version X
+ self.server.register_path(
+ "GET",
+ re.compile("^/pull/$"),
+ lambda request: handler.on_pull_request(
+ request.args["origin"][0],
+ request.args["v"]
+ )
+ )
+
+ # This is when someone asks for a data item for a given server
+ # data_id pair.
+ self.server.register_path(
+ "GET",
+ re.compile("^/pdu/([^/]*)/([^/]*)/$"),
+ lambda request, pdu_origin, pdu_id: handler.on_pdu_request(
+ pdu_origin, pdu_id
+ )
+ )
+
+ # This is when someone asks for all data for a given context.
+ self.server.register_path(
+ "GET",
+ re.compile("^/state/([^/]*)/$"),
+ lambda request, context: handler.on_context_state_request(
+ context
+ )
+ )
+
+ self.server.register_path(
+ "GET",
+ re.compile("^/paginate/([^/]*)/$"),
+ lambda request, context: self._on_paginate_request(
+ context, request.args["v"],
+ request.args["limit"]
+ )
+ )
+
+ self.server.register_path(
+ "GET",
+ re.compile("^/context/([^/]*)/$"),
+ lambda request, context: handler.on_context_pdus_request(context)
+ )
+
+ @defer.inlineCallbacks
+ @log_function
+ def _on_send_request(self, request, transaction_id):
+ """ Called on PUT /send/<transaction_id>/
+
+ Args:
+ request (twisted.web.http.Request): The HTTP request.
+ transaction_id (str): The transaction_id associated with this
+ request. This is *not* None.
+
+ Returns:
+ Deferred: Results in a tuple of `(code, response)`, where
+ `response` is a python dict to be converted into JSON that is
+ used as the response body.
+ """
+ # Parse the request
+ try:
+ data = request.content.read()
+
+ l = data[:20].encode("string_escape")
+ logger.debug("Got data: \"%s\"", l)
+
+ transaction_data = json.loads(data)
+
+ logger.debug(
+ "Decoded %s: %s",
+ transaction_id, str(transaction_data)
+ )
+
+ # We should ideally be getting this from the security layer.
+ # origin = body["origin"]
+
+ # Add some extra data to the transaction dict that isn't included
+ # in the request body.
+ transaction_data.update(
+ transaction_id=transaction_id,
+ destination=self.server_name
+ )
+
+ except Exception as e:
+ logger.exception(e)
+ defer.returnValue((400, {"error": "Invalid transaction"}))
+ return
+
+ code, response = yield self.received_handler.on_incoming_transaction(
+ transaction_data
+ )
+
+ defer.returnValue((code, response))
+
+ @defer.inlineCallbacks
+ @log_function
+ def _do_request_for_transaction(self, destination, path, args={}):
+ """
+ Args:
+ destination (str)
+ path (str)
+ args (dict): This is parsed directly to the HttpClient.
+
+ Returns:
+ Deferred: Results in a dict.
+ """
+
+ data = yield self.client.get_json(
+ destination,
+ path=path,
+ args=args,
+ )
+
+ # Add certain keys to the JSON, ready for decoding as a Transaction
+ data.update(
+ origin=destination,
+ destination=self.server_name,
+ transaction_id=None
+ )
+
+ defer.returnValue(data)
+
+ @log_function
+ def _on_paginate_request(self, context, v_list, limits):
+ if not limits:
+ return defer.succeed(
+ (400, {"error": "Did not include limit param"})
+ )
+
+ limit = int(limits[-1])
+
+ versions = [v.split(",", 1) for v in v_list]
+
+ return self.request_handler.on_paginate_request(
+ context, versions, limit)
+
+
+class TransportReceivedHandler(object):
+ """ Callbacks used when we receive a transaction
+ """
+ def on_incoming_transaction(self, transaction):
+ """ Called on PUT /send/<transaction_id>, or on response to a request
+ that we sent (e.g. a pagination request)
+
+ Args:
+ transaction (synapse.transaction.Transaction): The transaction that
+ was sent to us.
+
+ Returns:
+ twisted.internet.defer.Deferred: A deferred that get's fired when
+ the transaction has finished being processed.
+
+ The result should be a tuple in the form of
+ `(response_code, respond_body)`, where `response_body` is a python
+ dict that will get serialized to JSON.
+
+ On errors, the dict should have an `error` key with a brief message
+ of what went wrong.
+ """
+ pass
+
+
+class TransportRequestHandler(object):
+ """ Handlers used when someone want's data from us
+ """
+ def on_pull_request(self, versions):
+ """ Called on GET /pull/?v=...
+
+ This is hit when a remote home server wants to get all data
+ after a given transaction. Mainly used when a home server comes back
+ online and wants to get everything it has missed.
+
+ Args:
+ versions (list): A list of transaction_ids that should be used to
+ determine what PDUs the remote side have not yet seen.
+
+ Returns:
+ Deferred: Resultsin a tuple in the form of
+ `(response_code, respond_body)`, where `response_body` is a python
+ dict that will get serialized to JSON.
+
+ On errors, the dict should have an `error` key with a brief message
+ of what went wrong.
+ """
+ pass
+
+ def on_pdu_request(self, pdu_origin, pdu_id):
+ """ Called on GET /pdu/<pdu_origin>/<pdu_id>/
+
+ Someone wants a particular PDU. This PDU may or may not have originated
+ from us.
+
+ Args:
+ pdu_origin (str)
+ pdu_id (str)
+
+ Returns:
+ Deferred: Resultsin a tuple in the form of
+ `(response_code, respond_body)`, where `response_body` is a python
+ dict that will get serialized to JSON.
+
+ On errors, the dict should have an `error` key with a brief message
+ of what went wrong.
+ """
+ pass
+
+ def on_context_state_request(self, context):
+ """ Called on GET /state/<context>/
+
+ Get's hit when someone wants all the *current* state for a given
+ contexts.
+
+ Args:
+ context (str): The name of the context that we're interested in.
+
+ Returns:
+ twisted.internet.defer.Deferred: A deferred that get's fired when
+ the transaction has finished being processed.
+
+ The result should be a tuple in the form of
+ `(response_code, respond_body)`, where `response_body` is a python
+ dict that will get serialized to JSON.
+
+ On errors, the dict should have an `error` key with a brief message
+ of what went wrong.
+ """
+ pass
+
+ def on_paginate_request(self, context, versions, limit):
+ """ Called on GET /paginate/<context>/?v=...&limit=...
+
+ Get's hit when we want to paginate backwards on a given context from
+ the given point.
+
+ Args:
+ context (str): The context to paginate on
+ versions (list): A list of 2-tuple's representing where to paginate
+ from, in the form `(pdu_id, origin)`
+ limit (int): How many pdus to return.
+
+ Returns:
+ Deferred: Resultsin a tuple in the form of
+ `(response_code, respond_body)`, where `response_body` is a python
+ dict that will get serialized to JSON.
+
+ On errors, the dict should have an `error` key with a brief message
+ of what went wrong.
+ """
+ pass
|
