Implement presence event source. Change the way the notifier indexes listeners

3 files changed, 17 insertions, 12 deletions

diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 8c34776245..aabec37fc0 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -50,7 +50,12 @@ class EventStreamHandler(BaseHandler):
         if pagin_config.from_token is None:
             pagin_config.from_token = None
 
-        events, tokens = yield self.notifier.get_events_for(auth_user, pagin_config, timeout)
+        rm_handler = self.hs.get_handlers().room_member_handler
+        room_ids = yield rm_handler.get_rooms_for_user(auth_user)
+
+        events, tokens = yield self.notifier.get_events_for(
+            auth_user, room_ids, pagin_config, timeout
+        )
 
         chunks = [
             e.get_dict() if isinstance(e, SynapseEvent) else e
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 8408266da0..9a690258de 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -676,12 +676,7 @@ class PresenceHandler(BaseHandler):
         statuscache.make_event(user=observed_user, clock=self.clock)
 
         self.notifier.on_new_user_event(
-            observer_user.to_string(),
-            event_data=statuscache.make_event(
-                user=observed_user,
-                clock=self.clock
-            ),
-            store_id=statuscache.serial
+            [observer_user],
         )
 
 
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 6fbe84ea40..19ade10a91 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -120,8 +120,11 @@ class MessageHandler(BaseHandler):
         else:
             from_token = yield self.hs.get_event_sources().get_current_token()
 
+        user = self.hs.parse_userid(user_id)
+
         events, next_token = yield data_source.get_pagination_rows(
-            from_token, pagin_config.to_token, pagin_config.limit, room_id
+            user, from_token, pagin_config.to_token, pagin_config.limit,
+            room_id
         )
 
         chunk = {
@@ -265,6 +268,8 @@ class MessageHandler(BaseHandler):
             membership_list=[Membership.INVITE, Membership.JOIN]
         )
 
+        user = self.hs.parse_userid(user_id)
+
         rooms_ret = []
 
         # FIXME (erikj): We need to not generate this token,
@@ -272,8 +277,8 @@ class MessageHandler(BaseHandler):
 
         # FIXME (erikj): Fix this.
         presence_stream = self.hs.get_event_sources().sources[1]
-        presence = yield presence_stream.get_new_events_for_user(
-            user_id, now_token, None, None
+        presence, _ = yield presence_stream.get_pagination_rows(
+            user, now_token, None, None, None
         )
 
         limit = pagin_config.limit
@@ -297,7 +302,7 @@ class MessageHandler(BaseHandler):
                 messages, token = yield self.store.get_recent_events_for_room(
                     event.room_id,
                     limit=limit,
-                    end_token=now_token.events_key.to_string(),
+                    end_token=now_token.events_key,
                 )
 
                 d["messages"] = {
@@ -311,7 +316,7 @@ class MessageHandler(BaseHandler):
             except:
                 logger.exception("Failed to get snapshot")
 
-        ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token.to_string()}
+        ret = {"rooms": rooms_ret, "presence": presence, "end": now_token.to_string()}
 
         defer.returnValue(ret)