diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
index a4c29b484b..6620532a60 100644
--- a/synapse/federation/replication.py
+++ b/synapse/federation/replication.py
@@ -256,23 +256,21 @@ class ReplicationLayer(object):
@defer.inlineCallbacks
@log_function
- def get_state_for_context(self, destination, context, event_id):
- """Requests all of the `current` state PDUs for a given context from
+ def get_state_for_room(self, destination, room_id, event_id):
+ """Requests all of the `current` state PDUs for a given room from
a remote home server.
Args:
destination (str): The remote homeserver to query for the state.
- context (str): The context we're interested in.
+ room_id (str): The id of the room we're interested in.
event_id (str): The id of the event we want the state at.
Returns:
Deferred: Results in a list of PDUs.
"""
- result = yield self.transport_layer.get_context_state(
- destination,
- context,
- event_id=event_id,
+ result = yield self.transport_layer.get_room_state(
+ destination, room_id, event_id=event_id,
)
pdus = [
@@ -288,9 +286,9 @@ class ReplicationLayer(object):
@defer.inlineCallbacks
@log_function
- def get_event_auth(self, destination, context, event_id):
+ def get_event_auth(self, destination, room_id, event_id):
res = yield self.transport_layer.get_event_auth(
- destination, context, event_id,
+ destination, room_id, event_id,
)
auth_chain = [
@@ -304,9 +302,9 @@ class ReplicationLayer(object):
@defer.inlineCallbacks
@log_function
- def on_backfill_request(self, origin, context, versions, limit):
+ def on_backfill_request(self, origin, room_id, versions, limit):
pdus = yield self.handler.on_backfill_request(
- origin, context, versions, limit
+ origin, room_id, versions, limit
)
defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict()))
@@ -380,12 +378,10 @@ class ReplicationLayer(object):
@defer.inlineCallbacks
@log_function
- def on_context_state_request(self, origin, context, event_id):
+ def on_context_state_request(self, origin, room_id, event_id):
if event_id:
pdus = yield self.handler.get_state_for_pdu(
- origin,
- context,
- event_id,
+ origin, room_id, event_id,
)
auth_chain = yield self.store.get_auth_chain(
[pdu.event_id for pdu in pdus]
@@ -413,7 +409,7 @@ class ReplicationLayer(object):
@defer.inlineCallbacks
@log_function
def on_pull_request(self, origin, versions):
- raise NotImplementedError("Pull transacions not implemented")
+ raise NotImplementedError("Pull transactions not implemented")
@defer.inlineCallbacks
def on_query_request(self, query_type, args):
@@ -422,30 +418,21 @@ class ReplicationLayer(object):
defer.returnValue((200, response))
else:
defer.returnValue(
- (404, "No handler for Query type '%s'" % (query_type, ))
+ (404, "No handler for Query type '%s'" % (query_type,))
)
@defer.inlineCallbacks
- def on_make_join_request(self, context, user_id):
- pdu = yield self.handler.on_make_join_request(context, user_id)
+ def on_make_join_request(self, room_id, user_id):
+ pdu = yield self.handler.on_make_join_request(room_id, user_id)
time_now = self._clock.time_msec()
- defer.returnValue({
- "event": pdu.get_pdu_json(time_now),
- })
+ defer.returnValue({"event": pdu.get_pdu_json(time_now)})
@defer.inlineCallbacks
def on_invite_request(self, origin, content):
pdu = self.event_from_pdu_json(content)
ret_pdu = yield self.handler.on_invite_request(origin, pdu)
time_now = self._clock.time_msec()
- defer.returnValue(
- (
- 200,
- {
- "event": ret_pdu.get_pdu_json(time_now),
- }
- )
- )
+ defer.returnValue((200, {"event": ret_pdu.get_pdu_json(time_now)}))
@defer.inlineCallbacks
def on_send_join_request(self, origin, content):
@@ -462,26 +449,17 @@ class ReplicationLayer(object):
}))
@defer.inlineCallbacks
- def on_event_auth(self, origin, context, event_id):
+ def on_event_auth(self, origin, room_id, event_id):
time_now = self._clock.time_msec()
auth_pdus = yield self.handler.on_event_auth(event_id)
- defer.returnValue(
- (
- 200,
- {
- "auth_chain": [
- a.get_pdu_json(time_now) for a in auth_pdus
- ],
- }
- )
- )
+ defer.returnValue((200, {
+ "auth_chain": [a.get_pdu_json(time_now) for a in auth_pdus],
+ }))
@defer.inlineCallbacks
- def make_join(self, destination, context, user_id):
+ def make_join(self, destination, room_id, user_id):
ret = yield self.transport_layer.make_join(
- destination=destination,
- context=context,
- user_id=user_id,
+ destination, room_id, user_id
)
pdu_dict = ret["event"]
@@ -494,10 +472,10 @@ class ReplicationLayer(object):
def send_join(self, destination, pdu):
time_now = self._clock.time_msec()
_, content = yield self.transport_layer.send_join(
- destination,
- pdu.room_id,
- pdu.event_id,
- pdu.get_pdu_json(time_now),
+ destination=destination,
+ room_id=pdu.room_id,
+ event_id=pdu.event_id,
+ content=pdu.get_pdu_json(time_now),
)
logger.debug("Got content: %s", content)
@@ -507,9 +485,6 @@ class ReplicationLayer(object):
for p in content.get("state", [])
]
- # FIXME: We probably want to do something with the auth_chain given
- # to us
-
auth_chain = [
self.event_from_pdu_json(p, outlier=True)
for p in content.get("auth_chain", [])
@@ -523,11 +498,11 @@ class ReplicationLayer(object):
})
@defer.inlineCallbacks
- def send_invite(self, destination, context, event_id, pdu):
+ def send_invite(self, destination, room_id, event_id, pdu):
time_now = self._clock.time_msec()
code, content = yield self.transport_layer.send_invite(
destination=destination,
- context=context,
+ room_id=room_id,
event_id=event_id,
content=pdu.get_pdu_json(time_now),
)
@@ -657,7 +632,7 @@ class ReplicationLayer(object):
"_handle_new_pdu getting state for %s",
pdu.room_id
)
- state, auth_chain = yield self.get_state_for_context(
+ state, auth_chain = yield self.get_state_for_room(
origin, pdu.room_id, pdu.event_id,
)
@@ -816,7 +791,7 @@ class _TransactionQueue(object):
logger.info("TX [%s] is ready for retry", destination)
logger.info("TX [%s] _attempt_new_transaction", destination)
-
+
if destination in self.pending_transactions:
# XXX: pending_transactions can get stuck on by a never-ending
# request at which point pending_pdus_by_dest just keeps growing.
@@ -830,14 +805,15 @@ class _TransactionQueue(object):
pending_failures = self.pending_failures_by_dest.pop(destination, [])
if pending_pdus:
- logger.info("TX [%s] len(pending_pdus_by_dest[dest]) = %d", destination, len(pending_pdus))
+ logger.info("TX [%s] len(pending_pdus_by_dest[dest]) = %d",
+ destination, len(pending_pdus))
if not pending_pdus and not pending_edus and not pending_failures:
return
logger.debug(
- "TX [%s] Attempting new transaction "
- "(pdus: %d, edus: %d, failures: %d)",
+ "TX [%s] Attempting new transaction"
+ " (pdus: %d, edus: %d, failures: %d)",
destination,
len(pending_pdus),
len(pending_edus),
diff --git a/synapse/federation/transport/__init__.py b/synapse/federation/transport/__init__.py
new file mode 100644
index 0000000000..6800ac46c5
--- /dev/null
+++ b/synapse/federation/transport/__init__.py
@@ -0,0 +1,62 @@
+# -*- 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.
+
+"""The transport layer is responsible for both sending transactions to remote
+home servers and receiving a variety of requests from other home servers.
+
+By default this is done over HTTPS (and all home servers are required to
+support HTTPS), however individual pairings of servers may decide to
+communicate over a different (albeit still reliable) protocol.
+"""
+
+from .server import TransportLayerServer
+from .client import TransportLayerClient
+
+
+class TransportLayer(TransportLayerServer, TransportLayerClient):
+ """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, homeserver, 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.keyring = homeserver.get_keyring()
+ self.server_name = server_name
+ self.server = server
+ self.client = client
+ self.request_handler = None
+ self.received_handler = None
diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py
new file mode 100644
index 0000000000..e634a3a213
--- /dev/null
+++ b/synapse/federation/transport/client.py
@@ -0,0 +1,215 @@
+# -*- 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
+import json
+
+
+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()
+
+ code, 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: %d",
+ transaction.destination, transaction.transaction_id, code
+ )
+
+ defer.returnValue((code, response))
+
+ @defer.inlineCallbacks
+ @log_function
+ def make_query(self, destination, query_type, args, retry_on_dns_fail):
+ path = PREFIX + "/query/%s" % query_type
+
+ response = yield self.client.get_json(
+ destination=destination,
+ path=path,
+ args=args,
+ retry_on_dns_fail=retry_on_dns_fail,
+ )
+
+ defer.returnValue(response)
+
+ @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)
+
+ response = yield self.client.get_json(
+ destination=destination,
+ path=path,
+ retry_on_dns_fail=retry_on_dns_fail,
+ )
+
+ defer.returnValue(response)
+
+ @defer.inlineCallbacks
+ @log_function
+ def send_join(self, destination, room_id, event_id, content):
+ path = PREFIX + "/send_join/%s/%s" % (room_id, event_id)
+
+ code, content = yield self.client.put_json(
+ destination=destination,
+ path=path,
+ data=content,
+ )
+
+ if not 200 <= code < 300:
+ raise RuntimeError("Got %d from send_join", code)
+
+ defer.returnValue(json.loads(content))
+
+ @defer.inlineCallbacks
+ @log_function
+ def send_invite(self, destination, room_id, event_id, content):
+ path = PREFIX + "/invite/%s/%s" % (room_id, event_id)
+
+ code, content = yield self.client.put_json(
+ destination=destination,
+ path=path,
+ data=content,
+ )
+
+ if not 200 <= code < 300:
+ raise RuntimeError("Got %d from send_invite", code)
+
+ defer.returnValue(json.loads(content))
+
+ @defer.inlineCallbacks
+ @log_function
+ def get_event_auth(self, destination, room_id, event_id):
+ path = PREFIX + "/event_auth/%s/%s" % (room_id, event_id)
+
+ response = yield self.client.get_json(
+ destination=destination,
+ path=path,
+ )
+
+ defer.returnValue(response)
diff --git a/synapse/federation/transport.py b/synapse/federation/transport/server.py
index 1f0f06e0fe..a380a6910b 100644
--- a/synapse/federation/transport.py
+++ b/synapse/federation/transport/server.py
@@ -13,14 +13,6 @@
# 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.api.urls import FEDERATION_PREFIX as PREFIX
@@ -35,241 +27,8 @@ 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, homeserver, 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.keyring = homeserver.get_keyring()
- 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, event_id=None):
- """ 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)
-
- subpath = "/state/%s/" % context
-
- args = {}
- if event_id:
- args["event_id"] = event_id
-
- return self._do_request_for_transaction(
- destination, subpath, args=args
- )
-
- @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)
-
- subpath = "/event/%s/" % (event_id, )
-
- return self._do_request_for_transaction(destination, subpath)
-
- @log_function
- def backfill(self, dest, context, event_tuples, limit):
- """ Requests `limit` previous PDUs in a given context before list of
- PDUs.
-
- Args:
- dest (str)
- context (str)
- event_tuples (list)
- limt (int)
-
- Returns:
- Deferred: Results in a dict received from the remote homeserver.
- """
- logger.debug(
- "backfill dest=%s, context=%s, event_tuples=%s, limit=%s",
- dest, context, repr(event_tuples), str(limit)
- )
-
- if not event_tuples:
- # TODO: raise?
- return
-
- subpath = "/backfill/%s/" % (context,)
-
- args = {
- "v": event_tuples,
- "limit": [str(limit)],
- }
-
- return self._do_request_for_transaction(
- dest,
- subpath,
- 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()
-
- code, 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: %d",
- transaction.destination, transaction.transaction_id, code
- )
-
- defer.returnValue((code, response))
-
- @defer.inlineCallbacks
- @log_function
- def make_query(self, destination, query_type, args, retry_on_dns_fail):
- path = PREFIX + "/query/%s" % query_type
-
- response = yield self.client.get_json(
- destination=destination,
- path=path,
- args=args,
- retry_on_dns_fail=retry_on_dns_fail,
- )
-
- defer.returnValue(response)
-
- @defer.inlineCallbacks
- @log_function
- def make_join(self, destination, context, user_id, retry_on_dns_fail=True):
- path = PREFIX + "/make_join/%s/%s" % (context, user_id,)
-
- response = yield self.client.get_json(
- destination=destination,
- path=path,
- retry_on_dns_fail=retry_on_dns_fail,
- )
-
- defer.returnValue(response)
-
- @defer.inlineCallbacks
- @log_function
- def send_join(self, destination, context, event_id, content):
- path = PREFIX + "/send_join/%s/%s" % (
- context,
- event_id,
- )
-
- code, content = yield self.client.put_json(
- destination=destination,
- path=path,
- data=content,
- )
-
- if not 200 <= code < 300:
- raise RuntimeError("Got %d from send_join", code)
-
- defer.returnValue(json.loads(content))
-
- @defer.inlineCallbacks
- @log_function
- def send_invite(self, destination, context, event_id, content):
- path = PREFIX + "/invite/%s/%s" % (
- context,
- event_id,
- )
-
- code, content = yield self.client.put_json(
- destination=destination,
- path=path,
- data=content,
- )
-
- if not 200 <= code < 300:
- raise RuntimeError("Got %d from send_invite", code)
-
- defer.returnValue(json.loads(content))
-
- @defer.inlineCallbacks
- @log_function
- def get_event_auth(self, destination, context, event_id):
- path = PREFIX + "/event_auth/%s/%s" % (
- context,
- event_id,
- )
-
- response = yield self.client.get_json(
- destination=destination,
- path=path,
- )
-
- defer.returnValue(response)
+class TransportLayerServer(object):
+ """Handles incoming federation HTTP requests"""
@defer.inlineCallbacks
def _authenticate_request(self, request):
@@ -373,8 +132,6 @@ class TransportLayer(object):
"""
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",
@@ -528,34 +285,6 @@ class TransportLayer(object):
defer.returnValue((code, response))
- @defer.inlineCallbacks
- @log_function
- def _do_request_for_transaction(self, destination, subpath, 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=PREFIX + subpath,
- 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_backfill_request(self, origin, context, v_list, limits):
if not limits:
|
