diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index eef325a94b..6519f183df 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -29,6 +29,12 @@ logger = logging.getLogger(__name__)
class BaseHandler(object):
+ """
+ Common base class for the event handlers.
+
+ :type store: synapse.storage.events.StateStore
+ :type state_handler: synapse.state.StateHandler
+ """
def __init__(self, hs):
self.store = hs.get_datastore()
@@ -175,6 +181,8 @@ class BaseHandler(object):
if not suppress_auth:
self.auth.check(event, auth_events=context.current_state)
+ yield self.maybe_kick_guest_users(event, context.current_state.values())
+
if event.type == EventTypes.CanonicalAlias:
# Check the alias is acually valid (at this time at least)
room_alias_str = event.content.get("alias", None)
@@ -282,3 +290,58 @@ class BaseHandler(object):
federation_handler.handle_new_event(
event, destinations=destinations,
)
+
+ @defer.inlineCallbacks
+ def maybe_kick_guest_users(self, event, current_state):
+ # Technically this function invalidates current_state by changing it.
+ # Hopefully this isn't that important to the caller.
+ if event.type == EventTypes.GuestAccess:
+ guest_access = event.content.get("guest_access", "forbidden")
+ if guest_access != "can_join":
+ yield self.kick_guest_users(current_state)
+
+ @defer.inlineCallbacks
+ def kick_guest_users(self, current_state):
+ for member_event in current_state:
+ try:
+ if member_event.type != EventTypes.Member:
+ continue
+
+ if not self.hs.is_mine(UserID.from_string(member_event.state_key)):
+ continue
+
+ if member_event.content["membership"] not in {
+ Membership.JOIN,
+ Membership.INVITE
+ }:
+ continue
+
+ if (
+ "kind" not in member_event.content
+ or member_event.content["kind"] != "guest"
+ ):
+ continue
+
+ # We make the user choose to leave, rather than have the
+ # event-sender kick them. This is partially because we don't
+ # need to worry about power levels, and partially because guest
+ # users are a concept which doesn't hugely work over federation,
+ # and having homeservers have their own users leave keeps more
+ # of that decision-making and control local to the guest-having
+ # homeserver.
+ message_handler = self.hs.get_handlers().message_handler
+ yield message_handler.create_and_send_event(
+ {
+ "type": EventTypes.Member,
+ "state_key": member_event.state_key,
+ "content": {
+ "membership": Membership.LEAVE,
+ "kind": "guest"
+ },
+ "room_id": member_event.room_id,
+ "sender": member_event.state_key
+ },
+ ratelimit=False,
+ )
+ except Exception as e:
+ logger.warn("Error kicking guest user: %s" % (e,))
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 872051b8b9..c1bce07e31 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -564,7 +564,7 @@ class FederationHandler(BaseHandler):
@log_function
@defer.inlineCallbacks
- def do_invite_join(self, target_hosts, room_id, joinee):
+ def do_invite_join(self, target_hosts, room_id, joinee, content):
""" Attempts to join the `joinee` to the room `room_id` via the
server `target_host`.
@@ -584,7 +584,8 @@ class FederationHandler(BaseHandler):
target_hosts,
room_id,
joinee,
- "join"
+ "join",
+ content,
)
self.room_queues[room_id] = []
@@ -840,12 +841,14 @@ class FederationHandler(BaseHandler):
defer.returnValue(None)
@defer.inlineCallbacks
- def _make_and_verify_event(self, target_hosts, room_id, user_id, membership):
+ def _make_and_verify_event(self, target_hosts, room_id, user_id, membership,
+ content={},):
origin, pdu = yield self.replication_layer.make_membership_event(
target_hosts,
room_id,
user_id,
- membership
+ membership,
+ content,
)
logger.debug("Got response to make_%s: %s", membership, pdu)
@@ -1097,8 +1100,6 @@ class FederationHandler(BaseHandler):
context = yield self._prep_event(
origin, event,
state=state,
- backfilled=backfilled,
- current_state=current_state,
auth_events=auth_events,
)
@@ -1121,7 +1122,6 @@ class FederationHandler(BaseHandler):
origin,
ev_info["event"],
state=ev_info.get("state"),
- backfilled=backfilled,
auth_events=ev_info.get("auth_events"),
)
for ev_info in event_infos
@@ -1208,8 +1208,7 @@ class FederationHandler(BaseHandler):
defer.returnValue((event_stream_id, max_stream_id))
@defer.inlineCallbacks
- def _prep_event(self, origin, event, state=None, backfilled=False,
- current_state=None, auth_events=None):
+ def _prep_event(self, origin, event, state=None, auth_events=None):
outlier = event.internal_metadata.is_outlier()
context = yield self.state_handler.compute_event_context(
@@ -1242,6 +1241,10 @@ class FederationHandler(BaseHandler):
context.rejected = RejectedReason.AUTH_ERROR
+ if event.type == EventTypes.GuestAccess:
+ full_context = yield self.store.get_current_state(room_id=event.room_id)
+ yield self.maybe_kick_guest_users(event, full_context)
+
defer.returnValue(context)
@defer.inlineCallbacks
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 654ecd2b37..14051aee99 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -167,7 +167,7 @@ class MessageHandler(BaseHandler):
@defer.inlineCallbacks
def create_and_send_event(self, event_dict, ratelimit=True,
- token_id=None, txn_id=None):
+ token_id=None, txn_id=None, is_guest=False):
""" Given a dict from a client, create and handle a new event.
Creates an FrozenEvent object, filling out auth_events, prev_events,
@@ -213,7 +213,7 @@ class MessageHandler(BaseHandler):
if event.type == EventTypes.Member:
member_handler = self.hs.get_handlers().room_member_handler
- yield member_handler.change_membership(event, context)
+ yield member_handler.change_membership(event, context, is_guest=is_guest)
else:
yield self.handle_new_client_event(
event=event,
@@ -258,20 +258,30 @@ class MessageHandler(BaseHandler):
@defer.inlineCallbacks
def _check_in_room_or_world_readable(self, room_id, user_id, is_guest):
- if is_guest:
+ try:
+ # check_user_was_in_room will return the most recent membership
+ # event for the user if:
+ # * The user is a non-guest user, and was ever in the room
+ # * The user is a guest user, and has joined the room
+ # else it will throw.
+ member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
+ defer.returnValue((member_event.membership, member_event.event_id))
+ return
+ except AuthError, auth_error:
visibility = yield self.state_handler.get_current_state(
room_id, EventTypes.RoomHistoryVisibility, ""
)
- if visibility.content["history_visibility"] == "world_readable":
+ if (
+ visibility and
+ visibility.content["history_visibility"] == "world_readable"
+ ):
defer.returnValue((Membership.JOIN, None))
return
- else:
- raise AuthError(
- 403, "Guest access not allowed", errcode=Codes.GUEST_ACCESS_FORBIDDEN
- )
- else:
- member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
- defer.returnValue((member_event.membership, member_event.event_id))
+ if not is_guest:
+ raise auth_error
+ raise AuthError(
+ 403, "Guest access not allowed", errcode=Codes.GUEST_ACCESS_FORBIDDEN
+ )
@defer.inlineCallbacks
def get_state_events(self, user_id, room_id, is_guest=False):
@@ -456,7 +466,7 @@ class MessageHandler(BaseHandler):
defer.returnValue(ret)
@defer.inlineCallbacks
- def room_initial_sync(self, user_id, room_id, pagin_config=None):
+ def room_initial_sync(self, user_id, room_id, pagin_config=None, is_guest=False):
"""Capture the a snapshot of a room. If user is currently a member of
the room this will be what is currently in the room. If the user left
the room this will be what was in the room when they left.
@@ -473,15 +483,19 @@ class MessageHandler(BaseHandler):
A JSON serialisable dict with the snapshot of the room.
"""
- member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
+ membership, member_event_id = yield self._check_in_room_or_world_readable(
+ room_id,
+ user_id,
+ is_guest
+ )
- if member_event.membership == Membership.JOIN:
+ if membership == Membership.JOIN:
result = yield self._room_initial_sync_joined(
- user_id, room_id, pagin_config, member_event
+ user_id, room_id, pagin_config, membership, is_guest
)
- elif member_event.membership == Membership.LEAVE:
+ elif membership == Membership.LEAVE:
result = yield self._room_initial_sync_parted(
- user_id, room_id, pagin_config, member_event
+ user_id, room_id, pagin_config, membership, member_event_id, is_guest
)
private_user_data = []
@@ -497,19 +511,19 @@ class MessageHandler(BaseHandler):
@defer.inlineCallbacks
def _room_initial_sync_parted(self, user_id, room_id, pagin_config,
- member_event):
+ membership, member_event_id, is_guest):
room_state = yield self.store.get_state_for_events(
- [member_event.event_id], None
+ [member_event_id], None
)
- room_state = room_state[member_event.event_id]
+ room_state = room_state[member_event_id]
limit = pagin_config.limit if pagin_config else None
if limit is None:
limit = 10
stream_token = yield self.store.get_stream_token_for_event(
- member_event.event_id
+ member_event_id
)
messages, token = yield self.store.get_recent_events_for_room(
@@ -519,7 +533,7 @@ class MessageHandler(BaseHandler):
)
messages = yield self._filter_events_for_client(
- user_id, messages
+ user_id, messages, is_guest=is_guest
)
start_token = StreamToken(token[0], 0, 0, 0, 0)
@@ -528,7 +542,7 @@ class MessageHandler(BaseHandler):
time_now = self.clock.time_msec()
defer.returnValue({
- "membership": member_event.membership,
+ "membership": membership,
"room_id": room_id,
"messages": {
"chunk": [serialize_event(m, time_now) for m in messages],
@@ -542,7 +556,7 @@ class MessageHandler(BaseHandler):
@defer.inlineCallbacks
def _room_initial_sync_joined(self, user_id, room_id, pagin_config,
- member_event):
+ membership, is_guest):
current_state = yield self.state.get_current_state(
room_id=room_id,
)
@@ -574,12 +588,14 @@ class MessageHandler(BaseHandler):
@defer.inlineCallbacks
def get_presence():
- states = yield presence_handler.get_states(
- target_users=[UserID.from_string(m.user_id) for m in room_members],
- auth_user=auth_user,
- as_event=True,
- check_auth=False,
- )
+ states = {}
+ if not is_guest:
+ states = yield presence_handler.get_states(
+ target_users=[UserID.from_string(m.user_id) for m in room_members],
+ auth_user=auth_user,
+ as_event=True,
+ check_auth=False,
+ )
defer.returnValue(states.values())
@@ -599,7 +615,7 @@ class MessageHandler(BaseHandler):
).addErrback(unwrapFirstError)
messages = yield self._filter_events_for_client(
- user_id, messages
+ user_id, messages, is_guest=is_guest, require_all_visible_for_guests=False
)
start_token = now_token.copy_and_replace("room_key", token[0])
@@ -607,8 +623,7 @@ class MessageHandler(BaseHandler):
time_now = self.clock.time_msec()
- defer.returnValue({
- "membership": member_event.membership,
+ ret = {
"room_id": room_id,
"messages": {
"chunk": [serialize_event(m, time_now) for m in messages],
@@ -618,4 +633,8 @@ class MessageHandler(BaseHandler):
"state": state,
"presence": presence,
"receipts": receipts,
- })
+ }
+ if not is_guest:
+ ret["membership"] = membership
+
+ defer.returnValue(ret)
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 0b780cd528..aca65096fc 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -950,7 +950,8 @@ class PresenceHandler(BaseHandler):
)
while len(self._remote_offline_serials) > MAX_OFFLINE_SERIALS:
self._remote_offline_serials.pop() # remove the oldest
- del self._user_cachemap[user]
+ if user in self._user_cachemap:
+ del self._user_cachemap[user]
else:
# Remove the user from remote_offline_serials now that they're
# no longer offline
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 8cce8d0e99..3f04752581 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -369,7 +369,7 @@ class RoomMemberHandler(BaseHandler):
remotedomains.add(member.domain)
@defer.inlineCallbacks
- def change_membership(self, event, context, do_auth=True):
+ def change_membership(self, event, context, do_auth=True, is_guest=False):
""" Change the membership status of a user in a room.
Args:
@@ -390,6 +390,20 @@ class RoomMemberHandler(BaseHandler):
# if this HS is not currently in the room, i.e. we have to do the
# invite/join dance.
if event.membership == Membership.JOIN:
+ if is_guest:
+ guest_access = context.current_state.get(
+ (EventTypes.GuestAccess, ""),
+ None
+ )
+ is_guest_access_allowed = (
+ guest_access
+ and guest_access.content
+ and "guest_access" in guest_access.content
+ and guest_access.content["guest_access"] == "can_join"
+ )
+ if not is_guest_access_allowed:
+ raise AuthError(403, "Guest access not allowed")
+
yield self._do_join(event, context, do_auth=do_auth)
else:
if event.membership == Membership.LEAVE:
@@ -490,7 +504,8 @@ class RoomMemberHandler(BaseHandler):
yield handler.do_invite_join(
room_hosts,
room_id,
- event.user_id
+ event.user_id,
+ event.content,
)
else:
logger.debug("Doing normal join")
@@ -582,7 +597,6 @@ class RoomMemberHandler(BaseHandler):
medium,
address,
id_server,
- display_name,
token_id,
txn_id
):
@@ -609,7 +623,6 @@ class RoomMemberHandler(BaseHandler):
else:
yield self._make_and_store_3pid_invite(
id_server,
- display_name,
medium,
address,
room_id,
@@ -673,7 +686,6 @@ class RoomMemberHandler(BaseHandler):
def _make_and_store_3pid_invite(
self,
id_server,
- display_name,
medium,
address,
room_id,
@@ -681,7 +693,7 @@ class RoomMemberHandler(BaseHandler):
token_id,
txn_id
):
- token, public_key, key_validity_url = (
+ token, public_key, key_validity_url, display_name = (
yield self._ask_id_server_for_third_party_invite(
id_server,
medium,
@@ -725,10 +737,11 @@ class RoomMemberHandler(BaseHandler):
# TODO: Check for success
token = data["token"]
public_key = data["public_key"]
+ display_name = data["display_name"]
key_validity_url = "%s%s/_matrix/identity/api/v1/pubkey/isvalid" % (
id_server_scheme, id_server,
)
- defer.returnValue((token, public_key, key_validity_url))
+ defer.returnValue((token, public_key, key_validity_url, display_name))
class RoomListHandler(BaseHandler):
@@ -753,7 +766,7 @@ class RoomListHandler(BaseHandler):
class RoomContextHandler(BaseHandler):
@defer.inlineCallbacks
- def get_event_context(self, user, room_id, event_id, limit):
+ def get_event_context(self, user, room_id, event_id, limit, is_guest):
"""Retrieves events, pagination tokens and state around a given event
in a room.
@@ -777,11 +790,17 @@ class RoomContextHandler(BaseHandler):
)
results["events_before"] = yield self._filter_events_for_client(
- user.to_string(), results["events_before"]
+ user.to_string(),
+ results["events_before"],
+ is_guest=is_guest,
+ require_all_visible_for_guests=False
)
results["events_after"] = yield self._filter_events_for_client(
- user.to_string(), results["events_after"]
+ user.to_string(),
+ results["events_after"],
+ is_guest=is_guest,
+ require_all_visible_for_guests=False
)
if results["events_after"]:
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 5294d96466..6dc9d0fb92 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -47,9 +47,9 @@ class TimelineBatch(collections.namedtuple("TimelineBatch", [
class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [
- "room_id",
- "timeline",
- "state",
+ "room_id", # str
+ "timeline", # TimelineBatch
+ "state", # dict[(str, str), FrozenEvent]
"ephemeral",
"private_user_data",
])):
@@ -68,9 +68,9 @@ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [
class ArchivedSyncResult(collections.namedtuple("JoinedSyncResult", [
- "room_id",
- "timeline",
- "state",
+ "room_id", # str
+ "timeline", # TimelineBatch
+ "state", # dict[(str, str), FrozenEvent]
"private_user_data",
])):
__slots__ = []
@@ -87,8 +87,8 @@ class ArchivedSyncResult(collections.namedtuple("JoinedSyncResult", [
class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
- "room_id",
- "invite",
+ "room_id", # str
+ "invite", # FrozenEvent: the invite event
])):
__slots__ = []
@@ -254,15 +254,12 @@ class SyncHandler(BaseHandler):
room_id, sync_config, now_token, since_token=timeline_since_token
)
- current_state = yield self.state_handler.get_current_state(
- room_id
- )
- current_state_events = current_state.values()
+ current_state = yield self.get_state_at(room_id, now_token)
defer.returnValue(JoinedSyncResult(
room_id=room_id,
timeline=batch,
- state=current_state_events,
+ state=current_state,
ephemeral=ephemeral_by_room.get(room_id, []),
private_user_data=self.private_user_data_for_room(
room_id, tags_by_room
@@ -272,7 +269,7 @@ class SyncHandler(BaseHandler):
def private_user_data_for_room(self, room_id, tags_by_room):
private_user_data = []
tags = tags_by_room.get(room_id)
- if tags:
+ if tags is not None:
private_user_data.append({
"type": "m.tag",
"content": {"tags": tags},
@@ -311,8 +308,13 @@ class SyncHandler(BaseHandler):
ephemeral_by_room = {}
for event in typing:
- room_id = event.pop("room_id")
- ephemeral_by_room.setdefault(room_id, []).append(event)
+ # we want to exclude the room_id from the event, but modifying the
+ # result returned by the event source is poor form (it might cache
+ # the object)
+ room_id = event["room_id"]
+ event_copy = {k: v for (k, v) in event.iteritems()
+ if k != "room_id"}
+ ephemeral_by_room.setdefault(room_id, []).append(event_copy)
receipt_key = since_token.receipt_key if since_token else "0"
@@ -328,8 +330,11 @@ class SyncHandler(BaseHandler):
now_token = now_token.copy_and_replace("receipt_key", receipt_key)
for event in receipts:
- room_id = event.pop("room_id")
- ephemeral_by_room.setdefault(room_id, []).append(event)
+ room_id = event["room_id"]
+ # exclude room id, as above
+ event_copy = {k: v for (k, v) in event.iteritems()
+ if k != "room_id"}
+ ephemeral_by_room.setdefault(room_id, []).append(event_copy)
defer.returnValue((now_token, ephemeral_by_room))
@@ -346,14 +351,12 @@ class SyncHandler(BaseHandler):
room_id, sync_config, leave_token, since_token=timeline_since_token
)
- leave_state = yield self.store.get_state_for_events(
- [leave_event_id], None
- )
+ leave_state = yield self.store.get_state_for_event(leave_event_id)
defer.returnValue(ArchivedSyncResult(
room_id=room_id,
timeline=batch,
- state=leave_state[leave_event_id].values(),
+ state=leave_state,
private_user_data=self.private_user_data_for_room(
room_id, tags_by_room
),
@@ -417,6 +420,9 @@ class SyncHandler(BaseHandler):
if len(room_events) <= timeline_limit:
# There is no gap in any of the rooms. Therefore we can just
# partition the new events by room and return them.
+ logger.debug("Got %i events for incremental sync - not limited",
+ len(room_events))
+
invite_events = []
leave_events = []
events_by_room_id = {}
@@ -432,7 +438,12 @@ class SyncHandler(BaseHandler):
for room_id in joined_room_ids:
recents = events_by_room_id.get(room_id, [])
- state = [event for event in recents if event.is_state()]
+ logger.debug("Events for room %s: %r", room_id, recents)
+ state = {
+ (event.type, event.state_key): event
+ for event in recents if event.is_state()}
+ limited = False
+
if recents:
prev_batch = now_token.copy_and_replace(
"room_key", recents[0].internal_metadata.before
@@ -440,9 +451,13 @@ class SyncHandler(BaseHandler):
else:
prev_batch = now_token
- state, limited = yield self.check_joined_room(
- sync_config, room_id, state
- )
+ just_joined = yield self.check_joined_room(sync_config, state)
+ if just_joined:
+ logger.debug("User has just joined %s: needs full state",
+ room_id)
+ state = yield self.get_state_at(room_id, now_token)
+ # the timeline is inherently limited if we've just joined
+ limited = True
room_sync = JoinedSyncResult(
room_id=room_id,
@@ -457,10 +472,15 @@ class SyncHandler(BaseHandler):
room_id, tags_by_room
),
)
+ logger.debug("Result for room %s: %r", room_id, room_sync)
+
if room_sync:
joined.append(room_sync)
else:
+ logger.debug("Got %i events for incremental sync - hit limit",
+ len(room_events))
+
invite_events = yield self.store.get_invites_for_user(
sync_config.user.to_string()
)
@@ -499,6 +519,9 @@ class SyncHandler(BaseHandler):
@defer.inlineCallbacks
def load_filtered_recents(self, room_id, sync_config, now_token,
since_token=None):
+ """
+ :returns a Deferred TimelineBatch
+ """
limited = True
recents = []
filtering_factor = 2
@@ -550,6 +573,8 @@ class SyncHandler(BaseHandler):
Returns:
A Deferred JoinedSyncResult
"""
+ logger.debug("Doing incremental sync for room %s between %s and %s",
+ room_id, since_token, now_token)
# TODO(mjark): Check for redactions we might have missed.
@@ -559,31 +584,26 @@ class SyncHandler(BaseHandler):
logging.debug("Recents %r", batch)
- # TODO(mjark): This seems racy since this isn't being passed a
- # token to indicate what point in the stream this is
- current_state = yield self.state_handler.get_current_state(
- room_id
- )
- current_state_events = current_state.values()
+ current_state = yield self.get_state_at(room_id, now_token)
- state_at_previous_sync = yield self.get_state_at_previous_sync(
- room_id, since_token=since_token
+ state_at_previous_sync = yield self.get_state_at(
+ room_id, stream_position=since_token
)
- state_events_delta = yield self.compute_state_delta(
+ state = yield self.compute_state_delta(
since_token=since_token,
previous_state=state_at_previous_sync,
- current_state=current_state_events,
+ current_state=current_state,
)
- state_events_delta, _ = yield self.check_joined_room(
- sync_config, room_id, state_events_delta
- )
+ just_joined = yield self.check_joined_room(sync_config, state)
+ if just_joined:
+ state = yield self.get_state_at(room_id, now_token)
room_sync = JoinedSyncResult(
room_id=room_id,
timeline=batch,
- state=state_events_delta,
+ state=state,
ephemeral=ephemeral_by_room.get(room_id, []),
private_user_data=self.private_user_data_for_room(
room_id, tags_by_room
@@ -615,16 +635,12 @@ class SyncHandler(BaseHandler):
logging.debug("Recents %r", batch)
- # TODO(mjark): This seems racy since this isn't being passed a
- # token to indicate what point in the stream this is
- leave_state = yield self.store.get_state_for_events(
- [leave_event.event_id], None
+ state_events_at_leave = yield self.store.get_state_for_event(
+ leave_event.event_id
)
- state_events_at_leave = leave_state[leave_event.event_id].values()
-
- state_at_previous_sync = yield self.get_state_at_previous_sync(
- leave_event.room_id, since_token=since_token
+ state_at_previous_sync = yield self.get_state_at(
+ leave_event.room_id, stream_position=since_token
)
state_events_delta = yield self.compute_state_delta(
@@ -647,60 +663,77 @@ class SyncHandler(BaseHandler):
defer.returnValue(room_sync)
@defer.inlineCallbacks
- def get_state_at_previous_sync(self, room_id, since_token):
- """ Get the room state at the previous sync the client made.
- Returns:
- A Deferred list of Events.
+ def get_state_after_event(self, event):
+ """
+ Get the room state after the given event
+
+ :param synapse.events.EventBase event: event of interest
+ :return: A Deferred map from ((type, state_key)->Event)
+ """
+ state = yield self.store.get_state_for_event(event.event_id)
+ if event.is_state():
+ state = state.copy()
+ state[(event.type, event.state_key)] = event
+ defer.returnValue(state)
+
+ @defer.inlineCallbacks
+ def get_state_at(self, room_id, stream_position):
+ """ Get the room state at a particular stream position
+ :param str room_id: room for which to get state
+ :param StreamToken stream_position: point at which to get state
+ :returns: A Deferred map from ((type, state_key)->Event)
"""
last_events, token = yield self.store.get_recent_events_for_room(
- room_id, end_token=since_token.room_key, limit=1,
+ room_id, end_token=stream_position.room_key, limit=1,
)
if last_events:
- last_event = last_events[0]
- last_context = yield self.state_handler.compute_event_context(
- last_event
- )
- if last_event.is_state():
- state = [last_event] + last_context.current_state.values()
- else:
- state = last_context.current_state.values()
+ last_event = last_events[-1]
+ state = yield self.get_state_after_event(last_event)
+
else:
- state = ()
+ # no events in this room - so presumably no state
+ state = {}
defer.returnValue(state)
def compute_state_delta(self, since_token, previous_state, current_state):
""" Works out the differnce in state between the current state and the
state the client got when it last performed a sync.
- Returns:
- A list of events.
+
+ :param str since_token: the point we are comparing against
+ :param dict[(str,str), synapse.events.FrozenEvent] previous_state: the
+ state to compare to
+ :param dict[(str,str), synapse.events.FrozenEvent] current_state: the
+ new state
+
+ :returns A new event dictionary
"""
# TODO(mjark) Check if the state events were received by the server
# after the previous sync, since we need to include those state
# updates even if they occured logically before the previous event.
# TODO(mjark) Check for new redactions in the state events.
- previous_dict = {event.event_id: event for event in previous_state}
- state_delta = []
- for event in current_state:
- if event.event_id not in previous_dict:
- state_delta.append(event)
+
+ state_delta = {}
+ for key, event in current_state.iteritems():
+ if (key not in previous_state or
+ previous_state[key].event_id != event.event_id):
+ state_delta[key] = event
return state_delta
- @defer.inlineCallbacks
- def check_joined_room(self, sync_config, room_id, state_delta):
- joined = False
- limited = False
- for event in state_delta:
- if (
- event.type == EventTypes.Member
- and event.state_key == sync_config.user.to_string()
- ):
- if event.content["membership"] == Membership.JOIN:
- joined = True
-
- if joined:
- res = yield self.state_handler.get_current_state(room_id)
- state_delta = res.values()
- limited = True
+ def check_joined_room(self, sync_config, state_delta):
+ """
+ Check if the user has just joined the given room (so should
+ be given the full state)
- defer.returnValue((state_delta, limited))
+ :param sync_config:
+ :param dict[(str,str), synapse.events.FrozenEvent] state_delta: the
+ difference in state since the last sync
+
+ :returns A deferred Tuple (state_delta, limited)
+ """
+ join_event = state_delta.get((
+ EventTypes.Member, sync_config.user.to_string()), None)
+ if join_event is not None:
+ if join_event.content["membership"] == Membership.JOIN:
+ return True
+ return False
|
