Use new helper base class for membership requests
This commit is contained in:
parent
729b672823
commit
443da003bc
|
@ -20,16 +20,24 @@ from twisted.internet import defer
|
||||||
from synapse.api.errors import SynapseError
|
from synapse.api.errors import SynapseError
|
||||||
from synapse.handlers.room_member import RoomMemberHandler
|
from synapse.handlers.room_member import RoomMemberHandler
|
||||||
from synapse.replication.http.membership import (
|
from synapse.replication.http.membership import (
|
||||||
get_or_register_3pid_guest,
|
ReplicationRegister3PIDGuestRestServlet as Repl3PID,
|
||||||
notify_user_membership_change,
|
ReplicationRemoteJoinRestServlet as ReplRemoteJoin,
|
||||||
remote_join,
|
ReplicationRemoteRejectInviteRestServlet as ReplRejectInvite,
|
||||||
remote_reject_invite,
|
ReplicationUserJoinedLeftRoomRestServlet as ReplJoinedLeft,
|
||||||
)
|
)
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class RoomMemberWorkerHandler(RoomMemberHandler):
|
class RoomMemberWorkerHandler(RoomMemberHandler):
|
||||||
|
def __init__(self, hs):
|
||||||
|
super(RoomMemberWorkerHandler, self).__init__(hs)
|
||||||
|
|
||||||
|
self._get_register_3pid_client = Repl3PID.make_client(hs)
|
||||||
|
self._remote_join_client = ReplRemoteJoin.make_client(hs)
|
||||||
|
self._remote_reject_client = ReplRejectInvite.make_client(hs)
|
||||||
|
self._notify_change_client = ReplJoinedLeft.make_client(hs)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
|
def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
|
||||||
"""Implements RoomMemberHandler._remote_join
|
"""Implements RoomMemberHandler._remote_join
|
||||||
|
@ -37,10 +45,7 @@ class RoomMemberWorkerHandler(RoomMemberHandler):
|
||||||
if len(remote_room_hosts) == 0:
|
if len(remote_room_hosts) == 0:
|
||||||
raise SynapseError(404, "No known servers")
|
raise SynapseError(404, "No known servers")
|
||||||
|
|
||||||
ret = yield remote_join(
|
ret = yield self._remote_join_client(
|
||||||
self.simple_http_client,
|
|
||||||
host=self.config.worker_replication_host,
|
|
||||||
port=self.config.worker_replication_http_port,
|
|
||||||
requester=requester,
|
requester=requester,
|
||||||
remote_room_hosts=remote_room_hosts,
|
remote_room_hosts=remote_room_hosts,
|
||||||
room_id=room_id,
|
room_id=room_id,
|
||||||
|
@ -55,10 +60,7 @@ class RoomMemberWorkerHandler(RoomMemberHandler):
|
||||||
def _remote_reject_invite(self, requester, remote_room_hosts, room_id, target):
|
def _remote_reject_invite(self, requester, remote_room_hosts, room_id, target):
|
||||||
"""Implements RoomMemberHandler._remote_reject_invite
|
"""Implements RoomMemberHandler._remote_reject_invite
|
||||||
"""
|
"""
|
||||||
return remote_reject_invite(
|
return self._remote_reject_client(
|
||||||
self.simple_http_client,
|
|
||||||
host=self.config.worker_replication_host,
|
|
||||||
port=self.config.worker_replication_http_port,
|
|
||||||
requester=requester,
|
requester=requester,
|
||||||
remote_room_hosts=remote_room_hosts,
|
remote_room_hosts=remote_room_hosts,
|
||||||
room_id=room_id,
|
room_id=room_id,
|
||||||
|
@ -68,10 +70,7 @@ class RoomMemberWorkerHandler(RoomMemberHandler):
|
||||||
def _user_joined_room(self, target, room_id):
|
def _user_joined_room(self, target, room_id):
|
||||||
"""Implements RoomMemberHandler._user_joined_room
|
"""Implements RoomMemberHandler._user_joined_room
|
||||||
"""
|
"""
|
||||||
return notify_user_membership_change(
|
return self._notify_change_client(
|
||||||
self.simple_http_client,
|
|
||||||
host=self.config.worker_replication_host,
|
|
||||||
port=self.config.worker_replication_http_port,
|
|
||||||
user_id=target.to_string(),
|
user_id=target.to_string(),
|
||||||
room_id=room_id,
|
room_id=room_id,
|
||||||
change="joined",
|
change="joined",
|
||||||
|
@ -80,10 +79,7 @@ class RoomMemberWorkerHandler(RoomMemberHandler):
|
||||||
def _user_left_room(self, target, room_id):
|
def _user_left_room(self, target, room_id):
|
||||||
"""Implements RoomMemberHandler._user_left_room
|
"""Implements RoomMemberHandler._user_left_room
|
||||||
"""
|
"""
|
||||||
return notify_user_membership_change(
|
return self._notify_change_client(
|
||||||
self.simple_http_client,
|
|
||||||
host=self.config.worker_replication_host,
|
|
||||||
port=self.config.worker_replication_http_port,
|
|
||||||
user_id=target.to_string(),
|
user_id=target.to_string(),
|
||||||
room_id=room_id,
|
room_id=room_id,
|
||||||
change="left",
|
change="left",
|
||||||
|
@ -92,10 +88,7 @@ class RoomMemberWorkerHandler(RoomMemberHandler):
|
||||||
def get_or_register_3pid_guest(self, requester, medium, address, inviter_user_id):
|
def get_or_register_3pid_guest(self, requester, medium, address, inviter_user_id):
|
||||||
"""Implements RoomMemberHandler.get_or_register_3pid_guest
|
"""Implements RoomMemberHandler.get_or_register_3pid_guest
|
||||||
"""
|
"""
|
||||||
return get_or_register_3pid_guest(
|
return self._get_register_3pid_client(
|
||||||
self.simple_http_client,
|
|
||||||
host=self.config.worker_replication_host,
|
|
||||||
port=self.config.worker_replication_http_port,
|
|
||||||
requester=requester,
|
requester=requester,
|
||||||
medium=medium,
|
medium=medium,
|
||||||
address=address,
|
address=address,
|
||||||
|
|
|
@ -14,182 +14,53 @@
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import re
|
|
||||||
|
|
||||||
from twisted.internet import defer
|
from twisted.internet import defer
|
||||||
|
|
||||||
from synapse.api.errors import MatrixCodeMessageException, SynapseError
|
from synapse.http.servlet import parse_json_object_from_request
|
||||||
from synapse.http.servlet import RestServlet, parse_json_object_from_request
|
from synapse.replication.http._base import ReplicationEndpoint
|
||||||
from synapse.types import Requester, UserID
|
from synapse.types import Requester, UserID
|
||||||
from synapse.util.distributor import user_joined_room, user_left_room
|
from synapse.util.distributor import user_joined_room, user_left_room
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
class ReplicationRemoteJoinRestServlet(ReplicationEndpoint):
|
||||||
def remote_join(client, host, port, requester, remote_room_hosts,
|
"""Does a remote join for the given user to the given room
|
||||||
room_id, user_id, content):
|
|
||||||
"""Ask the master to do a remote join for the given user to the given room
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client (SimpleHttpClient)
|
|
||||||
host (str): host of master
|
|
||||||
port (int): port on master listening for HTTP replication
|
|
||||||
requester (Requester)
|
|
||||||
remote_room_hosts (list[str]): Servers to try and join via
|
|
||||||
room_id (str)
|
|
||||||
user_id (str)
|
|
||||||
content (dict): The event content to use for the join event
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Deferred
|
|
||||||
"""
|
|
||||||
uri = "http://%s:%s/_synapse/replication/remote_join" % (host, port)
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"requester": requester.serialize(),
|
|
||||||
"remote_room_hosts": remote_room_hosts,
|
|
||||||
"room_id": room_id,
|
|
||||||
"user_id": user_id,
|
|
||||||
"content": content,
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
result = yield client.post_json_get_json(uri, payload)
|
|
||||||
except MatrixCodeMessageException as e:
|
|
||||||
# We convert to SynapseError as we know that it was a SynapseError
|
|
||||||
# on the master process that we should send to the client. (And
|
|
||||||
# importantly, not stack traces everywhere)
|
|
||||||
raise SynapseError(e.code, e.msg, e.errcode)
|
|
||||||
defer.returnValue(result)
|
|
||||||
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
|
||||||
def remote_reject_invite(client, host, port, requester, remote_room_hosts,
|
|
||||||
room_id, user_id):
|
|
||||||
"""Ask master to reject the invite for the user and room.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client (SimpleHttpClient)
|
|
||||||
host (str): host of master
|
|
||||||
port (int): port on master listening for HTTP replication
|
|
||||||
requester (Requester)
|
|
||||||
remote_room_hosts (list[str]): Servers to try and reject via
|
|
||||||
room_id (str)
|
|
||||||
user_id (str)
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Deferred
|
|
||||||
"""
|
|
||||||
uri = "http://%s:%s/_synapse/replication/remote_reject_invite" % (host, port)
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"requester": requester.serialize(),
|
|
||||||
"remote_room_hosts": remote_room_hosts,
|
|
||||||
"room_id": room_id,
|
|
||||||
"user_id": user_id,
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
result = yield client.post_json_get_json(uri, payload)
|
|
||||||
except MatrixCodeMessageException as e:
|
|
||||||
# We convert to SynapseError as we know that it was a SynapseError
|
|
||||||
# on the master process that we should send to the client. (And
|
|
||||||
# importantly, not stack traces everywhere)
|
|
||||||
raise SynapseError(e.code, e.msg, e.errcode)
|
|
||||||
defer.returnValue(result)
|
|
||||||
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
|
||||||
def get_or_register_3pid_guest(client, host, port, requester,
|
|
||||||
medium, address, inviter_user_id):
|
|
||||||
"""Ask the master to get/create a guest account for given 3PID.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client (SimpleHttpClient)
|
|
||||||
host (str): host of master
|
|
||||||
port (int): port on master listening for HTTP replication
|
|
||||||
requester (Requester)
|
|
||||||
medium (str)
|
|
||||||
address (str)
|
|
||||||
inviter_user_id (str): The user ID who is trying to invite the
|
|
||||||
3PID
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Deferred[(str, str)]: A 2-tuple of `(user_id, access_token)` of the
|
|
||||||
3PID guest account.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
uri = "http://%s:%s/_synapse/replication/get_or_register_3pid_guest" % (host, port)
|
NAME = "remote_join"
|
||||||
|
PATH_ARGS = ("room_id", "user_id",)
|
||||||
payload = {
|
|
||||||
"requester": requester.serialize(),
|
|
||||||
"medium": medium,
|
|
||||||
"address": address,
|
|
||||||
"inviter_user_id": inviter_user_id,
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
result = yield client.post_json_get_json(uri, payload)
|
|
||||||
except MatrixCodeMessageException as e:
|
|
||||||
# We convert to SynapseError as we know that it was a SynapseError
|
|
||||||
# on the master process that we should send to the client. (And
|
|
||||||
# importantly, not stack traces everywhere)
|
|
||||||
raise SynapseError(e.code, e.msg, e.errcode)
|
|
||||||
defer.returnValue(result)
|
|
||||||
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
|
||||||
def notify_user_membership_change(client, host, port, user_id, room_id, change):
|
|
||||||
"""Notify master that a user has joined or left the room
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client (SimpleHttpClient)
|
|
||||||
host (str): host of master
|
|
||||||
port (int): port on master listening for HTTP replication.
|
|
||||||
user_id (str)
|
|
||||||
room_id (str)
|
|
||||||
change (str): Either "join" or "left"
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Deferred
|
|
||||||
"""
|
|
||||||
assert change in ("joined", "left")
|
|
||||||
|
|
||||||
uri = "http://%s:%s/_synapse/replication/user_%s_room" % (host, port, change)
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"user_id": user_id,
|
|
||||||
"room_id": room_id,
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
result = yield client.post_json_get_json(uri, payload)
|
|
||||||
except MatrixCodeMessageException as e:
|
|
||||||
# We convert to SynapseError as we know that it was a SynapseError
|
|
||||||
# on the master process that we should send to the client. (And
|
|
||||||
# importantly, not stack traces everywhere)
|
|
||||||
raise SynapseError(e.code, e.msg, e.errcode)
|
|
||||||
defer.returnValue(result)
|
|
||||||
|
|
||||||
|
|
||||||
class ReplicationRemoteJoinRestServlet(RestServlet):
|
|
||||||
PATTERNS = [re.compile("^/_synapse/replication/remote_join$")]
|
|
||||||
|
|
||||||
def __init__(self, hs):
|
def __init__(self, hs):
|
||||||
super(ReplicationRemoteJoinRestServlet, self).__init__()
|
super(ReplicationRemoteJoinRestServlet, self).__init__(hs)
|
||||||
|
|
||||||
self.federation_handler = hs.get_handlers().federation_handler
|
self.federation_handler = hs.get_handlers().federation_handler
|
||||||
self.store = hs.get_datastore()
|
self.store = hs.get_datastore()
|
||||||
self.clock = hs.get_clock()
|
self.clock = hs.get_clock()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _serialize_payload(requester, room_id, user_id, remote_room_hosts,
|
||||||
|
content):
|
||||||
|
"""
|
||||||
|
Args:
|
||||||
|
requester(Requester)
|
||||||
|
room_id (str)
|
||||||
|
user_id (str)
|
||||||
|
remote_room_hosts (list[str]): Servers to try and join via
|
||||||
|
content(dict): The event content to use for the join event
|
||||||
|
"""
|
||||||
|
return {
|
||||||
|
"requester": requester.serialize(),
|
||||||
|
"remote_room_hosts": remote_room_hosts,
|
||||||
|
"content": content,
|
||||||
|
}
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_POST(self, request):
|
def _handle_request(self, request, room_id, user_id):
|
||||||
content = parse_json_object_from_request(request)
|
content = parse_json_object_from_request(request)
|
||||||
|
|
||||||
remote_room_hosts = content["remote_room_hosts"]
|
remote_room_hosts = content["remote_room_hosts"]
|
||||||
room_id = content["room_id"]
|
|
||||||
user_id = content["user_id"]
|
|
||||||
event_content = content["content"]
|
event_content = content["content"]
|
||||||
|
|
||||||
requester = Requester.deserialize(self.store, content["requester"])
|
requester = Requester.deserialize(self.store, content["requester"])
|
||||||
|
@ -212,23 +83,39 @@ class ReplicationRemoteJoinRestServlet(RestServlet):
|
||||||
defer.returnValue((200, {}))
|
defer.returnValue((200, {}))
|
||||||
|
|
||||||
|
|
||||||
class ReplicationRemoteRejectInviteRestServlet(RestServlet):
|
class ReplicationRemoteRejectInviteRestServlet(ReplicationEndpoint):
|
||||||
PATTERNS = [re.compile("^/_synapse/replication/remote_reject_invite$")]
|
"""Rejects the invite for the user and room.
|
||||||
|
"""
|
||||||
|
|
||||||
|
NAME = "remote_reject_invite"
|
||||||
|
PATH_ARGS = ("room_id", "user_id",)
|
||||||
|
|
||||||
def __init__(self, hs):
|
def __init__(self, hs):
|
||||||
super(ReplicationRemoteRejectInviteRestServlet, self).__init__()
|
super(ReplicationRemoteRejectInviteRestServlet, self).__init__(hs)
|
||||||
|
|
||||||
self.federation_handler = hs.get_handlers().federation_handler
|
self.federation_handler = hs.get_handlers().federation_handler
|
||||||
self.store = hs.get_datastore()
|
self.store = hs.get_datastore()
|
||||||
self.clock = hs.get_clock()
|
self.clock = hs.get_clock()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _serialize_payload(requester, room_id, user_id, remote_room_hosts):
|
||||||
|
"""
|
||||||
|
Args:
|
||||||
|
requester(Requester)
|
||||||
|
room_id (str)
|
||||||
|
user_id (str)
|
||||||
|
remote_room_hosts (list[str]): Servers to try and reject via
|
||||||
|
"""
|
||||||
|
return {
|
||||||
|
"requester": requester.serialize(),
|
||||||
|
"remote_room_hosts": remote_room_hosts,
|
||||||
|
}
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_POST(self, request):
|
def _handle_request(self, request, room_id, user_id):
|
||||||
content = parse_json_object_from_request(request)
|
content = parse_json_object_from_request(request)
|
||||||
|
|
||||||
remote_room_hosts = content["remote_room_hosts"]
|
remote_room_hosts = content["remote_room_hosts"]
|
||||||
room_id = content["room_id"]
|
|
||||||
user_id = content["user_id"]
|
|
||||||
|
|
||||||
requester = Requester.deserialize(self.store, content["requester"])
|
requester = Requester.deserialize(self.store, content["requester"])
|
||||||
|
|
||||||
|
@ -264,18 +151,39 @@ class ReplicationRemoteRejectInviteRestServlet(RestServlet):
|
||||||
defer.returnValue((200, ret))
|
defer.returnValue((200, ret))
|
||||||
|
|
||||||
|
|
||||||
class ReplicationRegister3PIDGuestRestServlet(RestServlet):
|
class ReplicationRegister3PIDGuestRestServlet(ReplicationEndpoint):
|
||||||
PATTERNS = [re.compile("^/_synapse/replication/get_or_register_3pid_guest$")]
|
"""Gets/creates a guest account for given 3PID.
|
||||||
|
"""
|
||||||
|
|
||||||
|
NAME = "get_or_register_3pid_guest"
|
||||||
|
PATH_ARGS = ()
|
||||||
|
|
||||||
def __init__(self, hs):
|
def __init__(self, hs):
|
||||||
super(ReplicationRegister3PIDGuestRestServlet, self).__init__()
|
super(ReplicationRegister3PIDGuestRestServlet, self).__init__(hs)
|
||||||
|
|
||||||
self.registeration_handler = hs.get_handlers().registration_handler
|
self.registeration_handler = hs.get_handlers().registration_handler
|
||||||
self.store = hs.get_datastore()
|
self.store = hs.get_datastore()
|
||||||
self.clock = hs.get_clock()
|
self.clock = hs.get_clock()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _serialize_payload(requester, medium, address, inviter_user_id):
|
||||||
|
"""
|
||||||
|
Args:
|
||||||
|
requester(Requester)
|
||||||
|
medium (str)
|
||||||
|
address (str)
|
||||||
|
inviter_user_id (str): The user ID who is trying to invite the
|
||||||
|
3PID
|
||||||
|
"""
|
||||||
|
return {
|
||||||
|
"requester": requester.serialize(),
|
||||||
|
"medium": medium,
|
||||||
|
"address": address,
|
||||||
|
"inviter_user_id": inviter_user_id,
|
||||||
|
}
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_POST(self, request):
|
def _handle_request(self, request):
|
||||||
content = parse_json_object_from_request(request)
|
content = parse_json_object_from_request(request)
|
||||||
|
|
||||||
medium = content["medium"]
|
medium = content["medium"]
|
||||||
|
@ -296,23 +204,35 @@ class ReplicationRegister3PIDGuestRestServlet(RestServlet):
|
||||||
defer.returnValue((200, ret))
|
defer.returnValue((200, ret))
|
||||||
|
|
||||||
|
|
||||||
class ReplicationUserJoinedLeftRoomRestServlet(RestServlet):
|
class ReplicationUserJoinedLeftRoomRestServlet(ReplicationEndpoint):
|
||||||
PATTERNS = [re.compile("^/_synapse/replication/user_(?P<change>joined|left)_room$")]
|
"""Notifies that a user has joined or left the room
|
||||||
|
"""
|
||||||
|
|
||||||
|
NAME = "membership_change"
|
||||||
|
PATH_ARGS = ("room_id", "user_id", "change")
|
||||||
|
CACHE = False # No point caching as should return instantly.
|
||||||
|
|
||||||
def __init__(self, hs):
|
def __init__(self, hs):
|
||||||
super(ReplicationUserJoinedLeftRoomRestServlet, self).__init__()
|
super(ReplicationUserJoinedLeftRoomRestServlet, self).__init__(hs)
|
||||||
|
|
||||||
self.registeration_handler = hs.get_handlers().registration_handler
|
self.registeration_handler = hs.get_handlers().registration_handler
|
||||||
self.store = hs.get_datastore()
|
self.store = hs.get_datastore()
|
||||||
self.clock = hs.get_clock()
|
self.clock = hs.get_clock()
|
||||||
self.distributor = hs.get_distributor()
|
self.distributor = hs.get_distributor()
|
||||||
|
|
||||||
def on_POST(self, request, change):
|
@staticmethod
|
||||||
content = parse_json_object_from_request(request)
|
def _serialize_payload(room_id, user_id, change):
|
||||||
|
"""
|
||||||
|
Args:
|
||||||
|
room_id (str)
|
||||||
|
user_id (str)
|
||||||
|
change (str): Either "joined" or "left"
|
||||||
|
"""
|
||||||
|
assert change in ("joined", "left",)
|
||||||
|
|
||||||
user_id = content["user_id"]
|
return {}
|
||||||
room_id = content["room_id"]
|
|
||||||
|
|
||||||
|
def _handle_request(self, request, room_id, user_id, change):
|
||||||
logger.info("user membership change: %s in %s", user_id, room_id)
|
logger.info("user membership change: %s in %s", user_id, room_id)
|
||||||
|
|
||||||
user = UserID.from_string(user_id)
|
user = UserID.from_string(user_id)
|
||||||
|
|
Loading…
Reference in New Issue