diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 6cb756e476..b8e2c81969 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -164,6 +164,10 @@ class SyncHandler(BaseHandler):
"""
now_token = yield self.event_sources.get_current_token()
+ now_token, typing_by_room = yield self.typing_by_room(
+ sync_config, now_token
+ )
+
presence_stream = self.event_sources.sources["presence"]
# TODO (mjark): This looks wrong, shouldn't we be getting the presence
# UP to the present rather than after the present?
@@ -189,7 +193,7 @@ class SyncHandler(BaseHandler):
for event in room_list:
if event.membership == Membership.JOIN:
room_sync = yield self.initial_sync_for_joined_room(
- event.room_id, sync_config, now_token,
+ event.room_id, sync_config, now_token, typing_by_room
)
joined.append(room_sync)
elif event.membership == Membership.INVITE:
@@ -219,7 +223,8 @@ class SyncHandler(BaseHandler):
))
@defer.inlineCallbacks
- def initial_sync_for_joined_room(self, room_id, sync_config, now_token):
+ def initial_sync_for_joined_room(self, room_id, sync_config, now_token,
+ typing_by_room):
"""Sync a room for a client which is starting without any state
Returns:
A Deferred JoinedSyncResult.
@@ -238,10 +243,41 @@ class SyncHandler(BaseHandler):
room_id=room_id,
timeline=batch,
state=current_state_events,
- ephemeral=[],
+ ephemeral=typing_by_room.get(room_id, []),
))
@defer.inlineCallbacks
+ def typing_by_room(self, sync_config, now_token, since_token=None):
+ """Get the typing events for each room the user is in
+ Args:
+ sync_config (SyncConfig): The flags, filters and user for the sync.
+ now_token (StreamToken): Where the server is currently up to.
+ since_token (StreamToken): Where the server was when the client
+ last synced.
+ Returns:
+ A tuple of the now StreamToken, updated to reflect the which typing
+ events are included, and a dict mapping from room_id to a list of
+ typing events for that room.
+ """
+
+ typing_key = since_token.typing_key if since_token else "0"
+
+ typing_source = self.event_sources.sources["typing"]
+ typing, typing_key = yield typing_source.get_new_events_for_user(
+ user=sync_config.user,
+ from_key=typing_key,
+ limit=sync_config.filter.ephemeral_limit(),
+ )
+ now_token = now_token.copy_and_replace("typing_key", typing_key)
+
+ typing_by_room = {event["room_id"]: [event] for event in typing}
+ for event in typing:
+ event.pop("room_id")
+ logger.debug("Typing %r", typing_by_room)
+
+ defer.returnValue((now_token, typing_by_room))
+
+ @defer.inlineCallbacks
def initial_sync_for_archived_room(self, room_id, sync_config,
leave_event_id, leave_token):
"""Sync a room for a client which is starting without any state
@@ -280,18 +316,9 @@ class SyncHandler(BaseHandler):
)
now_token = now_token.copy_and_replace("presence_key", presence_key)
- typing_source = self.event_sources.sources["typing"]
- typing, typing_key = yield typing_source.get_new_events_for_user(
- user=sync_config.user,
- from_key=since_token.typing_key,
- limit=sync_config.filter.ephemeral_limit(),
+ now_token, typing_by_room = yield self.typing_by_room(
+ sync_config, now_token, since_token
)
- now_token = now_token.copy_and_replace("typing_key", typing_key)
-
- typing_by_room = {event["room_id"]: [event] for event in typing}
- for event in typing:
- event.pop("room_id")
- logger.debug("Typing %r", typing_by_room)
rm_handler = self.hs.get_handlers().room_member_handler
app_service = yield self.store.get_app_service_by_user_id(
|
