From 20d0db6cfb8efce079376eb6bd2c8cca4f4cab16 Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 17:09:15 +0100
Subject: Move the *EventSource classes into the handlers they relate to, so
 it's easier to find the code

---
 synapse/handlers/presence.py | 78 ++++++++++++++++++++++++++++++++++++++++++++
 synapse/handlers/room.py     | 48 +++++++++++++++++++++++++++
 2 files changed, 126 insertions(+)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 7731de85c0..1d3b02a9db 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -722,6 +722,84 @@ class PresenceHandler(BaseHandler):
         )
 
 
+class PresenceEventSource(object):
+    def __init__(self, hs):
+        self.hs = hs
+        self.clock = hs.get_clock()
+
+    def get_new_events_for_user(self, user, from_token, limit):
+        from_key = int(from_token.presence_key)
+
+        presence = self.hs.get_handlers().presence_handler
+        cachemap = presence._user_cachemap
+
+        # TODO(paul): limit, and filter by visibility
+        updates = [(k, cachemap[k]) for k in cachemap
+                   if from_key < cachemap[k].serial]
+
+        if updates:
+            clock = self.clock
+
+            latest_serial = max([x[1].serial for x in updates])
+            data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
+
+            end_token = from_token.copy_and_replace(
+                "presence_key", latest_serial
+            )
+            return ((data, end_token))
+        else:
+            end_token = from_token.copy_and_replace(
+                "presence_key", presence._user_cachemap_latest_serial
+            )
+            return (([], end_token))
+
+    def get_current_token_part(self):
+        presence = self.hs.get_handlers().presence_handler
+        return presence._user_cachemap_latest_serial
+
+    def get_pagination_rows(self, user, pagination_config, key):
+        # TODO (erikj): Does this make sense? Ordering?
+
+        from_token = pagination_config.from_token
+        to_token = pagination_config.to_token
+
+        from_key = int(from_token.presence_key)
+
+        if to_token:
+            to_key = int(to_token.presence_key)
+        else:
+            to_key = -1
+
+        presence = self.hs.get_handlers().presence_handler
+        cachemap = presence._user_cachemap
+
+        # TODO(paul): limit, and filter by visibility
+        updates = [(k, cachemap[k]) for k in cachemap
+                   if to_key < cachemap[k].serial < from_key]
+
+        if updates:
+            clock = self.clock
+
+            earliest_serial = max([x[1].serial for x in updates])
+            data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
+
+            if to_token:
+                next_token = to_token
+            else:
+                next_token = from_token
+
+            next_token = next_token.copy_and_replace(
+                "presence_key", earliest_serial
+            )
+            return ((data, next_token))
+        else:
+            if not to_token:
+                to_token = from_token.copy_and_replace(
+                    "presence_key", 0
+                )
+            return (([], to_token))
+
+
 class UserPresenceCache(object):
     """Store an observed user's state and status message.
 
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 3e41d7a46b..6fbb4bc18d 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -462,3 +462,51 @@ class RoomListHandler(BaseRoomHandler):
         chunk = yield self.store.get_rooms(is_public=True)
         # FIXME (erikj): START is no longer a valid value
         defer.returnValue({"start": "START", "end": "END", "chunk": chunk})
+
+
+class RoomEventSource(object):
+    def __init__(self, hs):
+        self.store = hs.get_datastore()
+
+    @defer.inlineCallbacks
+    def get_new_events_for_user(self, user, from_token, limit):
+        # We just ignore the key for now.
+
+        to_key = yield self.get_current_token_part()
+
+        events, end_key = yield self.store.get_room_events_stream(
+            user_id=user.to_string(),
+            from_key=from_token.events_key,
+            to_key=to_key,
+            room_id=None,
+            limit=limit,
+        )
+
+        end_token = from_token.copy_and_replace("events_key", end_key)
+
+        defer.returnValue((events, end_token))
+
+    def get_current_token_part(self):
+        return self.store.get_room_events_max_id()
+
+    @defer.inlineCallbacks
+    def get_pagination_rows(self, user, pagination_config, key):
+        from_token = pagination_config.from_token
+        to_token = pagination_config.to_token
+        limit = pagination_config.limit
+        direction = pagination_config.direction
+
+        to_key = to_token.events_key if to_token else None
+
+        events, next_key = yield self.store.paginate_room_events(
+            room_id=key,
+            from_key=from_token.events_key,
+            to_key=to_key,
+            direction=direction,
+            limit=limit,
+            with_feedback=True
+        )
+
+        next_token = from_token.copy_and_replace("events_key", next_key)
+
+        defer.returnValue((events, next_token))
-- 
cgit 1.4.1


From 6dd50da54ea944610eb3836621c45a2d6b2a532b Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 17:39:33 +0100
Subject: Define a new event stream data source for typing notifications
 (currently null)

---
 synapse/handlers/typing.py  | 14 ++++++++++++++
 synapse/streams/events.py   | 20 +++++++++++++-------
 synapse/types.py            |  2 +-
 tests/rest/test_presence.py |  8 +++++---
 4 files changed, 33 insertions(+), 11 deletions(-)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 9fab0ff37c..8a1e3dc5e8 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -145,3 +145,17 @@ class TypingNotificationHandler(BaseHandler):
             typing):
         # TODO(paul) steal this from presence.py
         pass
+
+
+class TypingNotificationEventSource(object):
+    def __init__(self, hs):
+        self.hs = hs
+
+    def get_new_events_for_user(self, user, from_token, limit):
+        return ([], 0)
+
+    def get_current_token_part(self):
+        return 0
+
+    def get_pagination_rows(self, user, pagination_config, key):
+        return ([], 0)
diff --git a/synapse/streams/events.py b/synapse/streams/events.py
index 321faf4b03..4bec6605bd 100644
--- a/synapse/streams/events.py
+++ b/synapse/streams/events.py
@@ -19,6 +19,7 @@ from synapse.types import StreamToken
 
 from synapse.handlers.presence import PresenceEventSource
 from synapse.handlers.room import RoomEventSource
+from synapse.handlers.typing import TypingNotificationEventSource
 
 
 class NullSource(object):
@@ -41,6 +42,7 @@ class EventSources(object):
     SOURCE_TYPES = {
         "room": RoomEventSource,
         "presence": PresenceEventSource,
+        "typing": TypingNotificationEventSource,
     }
 
     def __init__(self, hs):
@@ -49,15 +51,19 @@ class EventSources(object):
             for name, cls in EventSources.SOURCE_TYPES.items()
         }
 
-    @staticmethod
-    def create_token(events_key, presence_key):
-        return StreamToken(events_key=events_key, presence_key=presence_key)
-
     @defer.inlineCallbacks
     def get_current_token(self):
-        events_key = yield self.sources["room"].get_current_token_part()
-        presence_key = yield self.sources["presence"].get_current_token_part()
-        token = EventSources.create_token(events_key, presence_key)
+        token = StreamToken(
+            events_key=(
+                yield self.sources["room"].get_current_token_part()
+            ),
+            presence_key=(
+                yield self.sources["presence"].get_current_token_part()
+            ),
+            typing_key=(
+                yield self.sources["typing"].get_current_token_part()
+            )
+        )
         defer.returnValue(token)
 
 
diff --git a/synapse/types.py b/synapse/types.py
index aa6f589a20..d93b02a56e 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -97,7 +97,7 @@ class RoomID(DomainSpecificString):
 class StreamToken(
     namedtuple(
         "Token",
-        ("events_key", "presence_key")
+        ("events_key", "presence_key", "typing_key")
     )
 ):
     _SEPARATOR = "_"
diff --git a/tests/rest/test_presence.py b/tests/rest/test_presence.py
index 0f5fc21432..ab0d580c7c 100644
--- a/tests/rest/test_presence.py
+++ b/tests/rest/test_presence.py
@@ -295,7 +295,9 @@ class PresenceEventStreamTestCase(unittest.TestCase):
         # all be ours
 
         # I'll already get my own presence state change
-        self.assertEquals({"start": "0_1", "end": "0_1", "chunk": []}, response)
+        self.assertEquals({"start": "0_1_0", "end": "0_1_0", "chunk": []},
+            response
+        )
 
         self.mock_datastore.set_presence_state.return_value = defer.succeed(
                 {"state": ONLINE})
@@ -306,10 +308,10 @@ class PresenceEventStreamTestCase(unittest.TestCase):
                 state={"state": ONLINE})
 
         (code, response) = yield self.mock_resource.trigger("GET",
-                "/events?from=0_1&timeout=0", None)
+                "/events?from=0_1_0&timeout=0", None)
 
         self.assertEquals(200, code)
-        self.assertEquals({"start": "0_1", "end": "0_2", "chunk": [
+        self.assertEquals({"start": "0_1_0", "end": "0_2_0", "chunk": [
             {"type": "m.presence",
              "content": {
                  "user_id": "@banana:test",
-- 
cgit 1.4.1


From 4bfdec1eb28aa272391607b2cf1f24c781d9ba74 Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 18:39:09 +0100
Subject: Rename 'events_key' to 'room_key' so it matches the name of the event
 source

---
 synapse/handlers/message.py |  6 +++---
 synapse/handlers/room.py    | 10 +++++-----
 synapse/streams/events.py   |  2 +-
 synapse/types.py            |  2 +-
 4 files changed, 10 insertions(+), 10 deletions(-)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 3d7f97bcff..c8ff34e5f5 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -274,11 +274,11 @@ class MessageHandler(BaseRoomHandler):
                 messages, token = yield self.store.get_recent_events_for_room(
                     event.room_id,
                     limit=limit,
-                    end_token=now_token.events_key,
+                    end_token=now_token.room_key,
                 )
 
-                start_token = now_token.copy_and_replace("events_key", token[0])
-                end_token = now_token.copy_and_replace("events_key", token[1])
+                start_token = now_token.copy_and_replace("room_key", token[0])
+                end_token = now_token.copy_and_replace("room_key", token[1])
 
                 d["messages"] = {
                     "chunk": [m.get_dict() for m in messages],
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 6fbb4bc18d..b27bdecd43 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -476,13 +476,13 @@ class RoomEventSource(object):
 
         events, end_key = yield self.store.get_room_events_stream(
             user_id=user.to_string(),
-            from_key=from_token.events_key,
+            from_key=from_token.room_key,
             to_key=to_key,
             room_id=None,
             limit=limit,
         )
 
-        end_token = from_token.copy_and_replace("events_key", end_key)
+        end_token = from_token.copy_and_replace("room_key", end_key)
 
         defer.returnValue((events, end_token))
 
@@ -496,17 +496,17 @@ class RoomEventSource(object):
         limit = pagination_config.limit
         direction = pagination_config.direction
 
-        to_key = to_token.events_key if to_token else None
+        to_key = to_token.room_key if to_token else None
 
         events, next_key = yield self.store.paginate_room_events(
             room_id=key,
-            from_key=from_token.events_key,
+            from_key=from_token.room_key,
             to_key=to_key,
             direction=direction,
             limit=limit,
             with_feedback=True
         )
 
-        next_token = from_token.copy_and_replace("events_key", next_key)
+        next_token = from_token.copy_and_replace("room_key", next_key)
 
         defer.returnValue((events, next_token))
diff --git a/synapse/streams/events.py b/synapse/streams/events.py
index 4bec6605bd..8480368673 100644
--- a/synapse/streams/events.py
+++ b/synapse/streams/events.py
@@ -54,7 +54,7 @@ class EventSources(object):
     @defer.inlineCallbacks
     def get_current_token(self):
         token = StreamToken(
-            events_key=(
+            room_key=(
                 yield self.sources["room"].get_current_token_part()
             ),
             presence_key=(
diff --git a/synapse/types.py b/synapse/types.py
index d93b02a56e..1a9dceabf5 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -97,7 +97,7 @@ class RoomID(DomainSpecificString):
 class StreamToken(
     namedtuple(
         "Token",
-        ("events_key", "presence_key", "typing_key")
+        ("room_key", "presence_key", "typing_key")
     )
 ):
     _SEPARATOR = "_"
-- 
cgit 1.4.1


From 6797c7f1b138c7db59d8a7134091e0806cc44f74 Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 19:00:45 +0100
Subject: TypingNotificationEventSource has to return proper tokens, not int 0

---
 synapse/handlers/typing.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 8a1e3dc5e8..ecb9318d1c 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -152,10 +152,10 @@ class TypingNotificationEventSource(object):
         self.hs = hs
 
     def get_new_events_for_user(self, user, from_token, limit):
-        return ([], 0)
+        return ([], from_token)
 
     def get_current_token_part(self):
         return 0
 
     def get_pagination_rows(self, user, pagination_config, key):
-        return ([], 0)
+        return ([], pagination_config.from_token)
-- 
cgit 1.4.1


From eec67a675f7ea3545bfba79c6b753f63f7fd9b3b Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 19:13:55 +0100
Subject: Have EventSource's get_new_events_for_user() API work only on keys
 within that source, not overall eventstream tokens

---
 synapse/handlers/presence.py | 14 ++++----------
 synapse/handlers/room.py     |  8 +++-----
 synapse/handlers/typing.py   |  4 ++--
 synapse/notifier.py          | 30 ++++++++++++++++++++----------
 synapse/streams/events.py    |  7 ++++---
 5 files changed, 33 insertions(+), 30 deletions(-)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 1d3b02a9db..05bf145240 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -727,8 +727,8 @@ class PresenceEventSource(object):
         self.hs = hs
         self.clock = hs.get_clock()
 
-    def get_new_events_for_user(self, user, from_token, limit):
-        from_key = int(from_token.presence_key)
+    def get_new_events_for_user(self, user, from_key, limit):
+        from_key = int(from_key)
 
         presence = self.hs.get_handlers().presence_handler
         cachemap = presence._user_cachemap
@@ -743,15 +743,9 @@ class PresenceEventSource(object):
             latest_serial = max([x[1].serial for x in updates])
             data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
 
-            end_token = from_token.copy_and_replace(
-                "presence_key", latest_serial
-            )
-            return ((data, end_token))
+            return ((data, latest_serial))
         else:
-            end_token = from_token.copy_and_replace(
-                "presence_key", presence._user_cachemap_latest_serial
-            )
-            return (([], end_token))
+            return (([], presence._user_cachemap_latest_serial))
 
     def get_current_token_part(self):
         presence = self.hs.get_handlers().presence_handler
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index b27bdecd43..ce15420bf4 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -469,22 +469,20 @@ class RoomEventSource(object):
         self.store = hs.get_datastore()
 
     @defer.inlineCallbacks
-    def get_new_events_for_user(self, user, from_token, limit):
+    def get_new_events_for_user(self, user, from_key, limit):
         # We just ignore the key for now.
 
         to_key = yield self.get_current_token_part()
 
         events, end_key = yield self.store.get_room_events_stream(
             user_id=user.to_string(),
-            from_key=from_token.room_key,
+            from_key=from_key,
             to_key=to_key,
             room_id=None,
             limit=limit,
         )
 
-        end_token = from_token.copy_and_replace("room_key", end_key)
-
-        defer.returnValue((events, end_token))
+        defer.returnValue((events, end_key))
 
     def get_current_token_part(self):
         return self.store.get_room_events_max_id()
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index ecb9318d1c..238b063483 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -151,8 +151,8 @@ class TypingNotificationEventSource(object):
     def __init__(self, hs):
         self.hs = hs
 
-    def get_new_events_for_user(self, user, from_token, limit):
-        return ([], from_token)
+    def get_new_events_for_user(self, user, from_key, limit):
+        return ([], from_key)
 
     def get_current_token_part(self):
         return 0
diff --git a/synapse/notifier.py b/synapse/notifier.py
index b6d5ec4820..cb544e9886 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -95,7 +95,7 @@ class Notifier(object):
         """
         room_id = event.room_id
 
-        source = self.event_sources.sources["room"]
+        room_source = self.event_sources.sources["room"]
 
         listeners = self.rooms_to_listeners.get(room_id, set()).copy()
 
@@ -107,13 +107,17 @@ class Notifier(object):
         # TODO (erikj): Can we make this more efficient by hitting the
         # db once?
         for listener in listeners:
-            events, end_token = yield source.get_new_events_for_user(
+            events, end_key = yield room_source.get_new_events_for_user(
                 listener.user,
-                listener.from_token,
+                listener.from_token.room_key,
                 listener.limit,
             )
 
             if events:
+                end_token = listener.from_token.copy_and_replace(
+                    "room_key", end_key
+                )
+
                 listener.notify(
                     self, events, listener.from_token, end_token
                 )
@@ -126,7 +130,7 @@ class Notifier(object):
 
         Will wake up all listeners for the given users and rooms.
         """
-        source = self.event_sources.sources["presence"]
+        presence_source = self.event_sources.sources["presence"]
 
         listeners = set()
 
@@ -137,13 +141,17 @@ class Notifier(object):
             listeners |= self.rooms_to_listeners.get(room, set()).copy()
 
         for listener in listeners:
-            events, end_token = yield source.get_new_events_for_user(
+            events, end_key = yield presence_source.get_new_events_for_user(
                 listener.user,
-                listener.from_token,
+                listener.from_token.presence_key,
                 listener.limit,
             )
 
             if events:
+                end_token = listener.from_token.copy_and_replace(
+                    "presence_key", end_key
+                )
+
                 listener.notify(
                     self, events, listener.from_token, end_token
                 )
@@ -216,16 +224,18 @@ class Notifier(object):
         limit = listener.limit
 
         # TODO (erikj): DeferredList?
-        for source in self.event_sources.sources.values():
-            stuff, new_token = yield source.get_new_events_for_user(
+        for name, source in self.event_sources.sources.items():
+            keyname = "%s_key" % name
+
+            stuff, new_key = yield source.get_new_events_for_user(
                 listener.user,
-                from_token,
+                getattr(from_token, keyname),
                 limit,
             )
 
             events.extend(stuff)
 
-            from_token = new_token
+            from_token = from_token.copy_and_replace(keyname, new_key)
 
         end_token = from_token
 
diff --git a/synapse/streams/events.py b/synapse/streams/events.py
index 8480368673..43b6b1eba3 100644
--- a/synapse/streams/events.py
+++ b/synapse/streams/events.py
@@ -28,8 +28,8 @@ class NullSource(object):
     def __init__(self, hs):
         pass
 
-    def get_new_events_for_user(self, user, from_token, limit):
-        return defer.succeed(([], from_token))
+    def get_new_events_for_user(self, user, from_key, limit):
+        return defer.succeed(([], from_key))
 
     def get_current_token_part(self):
         return defer.succeed(0)
@@ -68,7 +68,8 @@ class EventSources(object):
 
 
 class StreamSource(object):
-    def get_new_events_for_user(self, user, from_token, limit):
+    def get_new_events_for_user(self, user, from_key, limit):
+        """from_key is the key within this event source."""
         raise NotImplementedError("get_new_events_for_user")
 
     def get_current_token_part(self):
-- 
cgit 1.4.1


From a8e8d1d06c078de49711768357267cf4168999ea Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 19:15:23 +0100
Subject: Renamed get_current_token_part to get_current_key

---
 synapse/handlers/presence.py |  2 +-
 synapse/handlers/room.py     |  4 ++--
 synapse/handlers/typing.py   |  2 +-
 synapse/streams/events.py    | 12 ++++++------
 4 files changed, 10 insertions(+), 10 deletions(-)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 05bf145240..cc28151e35 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -747,7 +747,7 @@ class PresenceEventSource(object):
         else:
             return (([], presence._user_cachemap_latest_serial))
 
-    def get_current_token_part(self):
+    def get_current_key(self):
         presence = self.hs.get_handlers().presence_handler
         return presence._user_cachemap_latest_serial
 
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index ce15420bf4..c54e0f963b 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -472,7 +472,7 @@ class RoomEventSource(object):
     def get_new_events_for_user(self, user, from_key, limit):
         # We just ignore the key for now.
 
-        to_key = yield self.get_current_token_part()
+        to_key = yield self.get_current_key()
 
         events, end_key = yield self.store.get_room_events_stream(
             user_id=user.to_string(),
@@ -484,7 +484,7 @@ class RoomEventSource(object):
 
         defer.returnValue((events, end_key))
 
-    def get_current_token_part(self):
+    def get_current_key(self):
         return self.store.get_room_events_max_id()
 
     @defer.inlineCallbacks
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 238b063483..3268427ecd 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -154,7 +154,7 @@ class TypingNotificationEventSource(object):
     def get_new_events_for_user(self, user, from_key, limit):
         return ([], from_key)
 
-    def get_current_token_part(self):
+    def get_current_key(self):
         return 0
 
     def get_pagination_rows(self, user, pagination_config, key):
diff --git a/synapse/streams/events.py b/synapse/streams/events.py
index 43b6b1eba3..08d6e6f733 100644
--- a/synapse/streams/events.py
+++ b/synapse/streams/events.py
@@ -31,7 +31,7 @@ class NullSource(object):
     def get_new_events_for_user(self, user, from_key, limit):
         return defer.succeed(([], from_key))
 
-    def get_current_token_part(self):
+    def get_current_key(self):
         return defer.succeed(0)
 
     def get_pagination_rows(self, user, pagination_config, key):
@@ -55,13 +55,13 @@ class EventSources(object):
     def get_current_token(self):
         token = StreamToken(
             room_key=(
-                yield self.sources["room"].get_current_token_part()
+                yield self.sources["room"].get_current_key()
             ),
             presence_key=(
-                yield self.sources["presence"].get_current_token_part()
+                yield self.sources["presence"].get_current_key()
             ),
             typing_key=(
-                yield self.sources["typing"].get_current_token_part()
+                yield self.sources["typing"].get_current_key()
             )
         )
         defer.returnValue(token)
@@ -72,8 +72,8 @@ class StreamSource(object):
         """from_key is the key within this event source."""
         raise NotImplementedError("get_new_events_for_user")
 
-    def get_current_token_part(self):
-        raise NotImplementedError("get_current_token_part")
+    def get_current_key(self):
+        raise NotImplementedError("get_current_key")
 
     def get_pagination_rows(self, user, pagination_config, key):
         raise NotImplementedError("get_rows")
-- 
cgit 1.4.1


From 93407cf7cff05fa99709ba5055bce329524cef32 Mon Sep 17 00:00:00 2001
From: "Paul \"LeoNerd\" Evans" <paul@matrix.org>
Date: Fri, 29 Aug 2014 19:53:33 +0100
Subject: Bugfixes on presence pushes on user joining:

 * No need to inform clients of status of remote users; as that will
   arrive in due course anyway. We don't -have- the state currently, so
   we'd only send an unknown message

 * Remember to bump the presence serial for the event source, so the
   notifiers will wake up and report it
---
 synapse/handlers/presence.py    |  17 +++---
 tests/handlers/test_presence.py | 128 ++++++++++++++++++++++------------------
 2 files changed, 79 insertions(+), 66 deletions(-)

(limited to 'synapse/handlers')

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index cc28151e35..93bd07b196 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -260,19 +260,18 @@ class PresenceHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def user_joined_room(self, user, room_id):
-
         if user.is_mine:
-            self.push_update_to_local_and_remote(
-                observed_user=user,
-                room_ids=[room_id],
-                statuscache=self._get_or_offline_usercache(user),
-            )
+            statuscache = self._get_or_make_usercache(user)
 
-        else:
-            self.push_update_to_clients(
+            # No actual update but we need to bump the serial anyway for the
+            # event source
+            self._user_cachemap_latest_serial += 1
+            statuscache.update({}, serial=self._user_cachemap_latest_serial)
+
+            self.push_update_to_local_and_remote(
                 observed_user=user,
                 room_ids=[room_id],
-                statuscache=self._get_or_offline_usercache(user),
+                statuscache=statuscache,
             )
 
         # We also want to tell them about current presence of people.
diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py
index fcd7a784cd..0a176bdd44 100644
--- a/tests/handlers/test_presence.py
+++ b/tests/handlers/test_presence.py
@@ -514,13 +514,6 @@ class PresencePushTestCase(unittest.TestCase):
             )
         hs.handlers = JustPresenceHandlers(hs)
 
-        def update(*args,**kwargs):
-            # print "mock_update_client: Args=%s, kwargs=%s" %(args, kwargs,)
-            return defer.succeed(None)
-
-        self.mock_update_client = Mock()
-        self.mock_update_client.side_effect = update
-
         self.datastore = hs.get_datastore()
 
         def get_received_txn_response(*args):
@@ -528,7 +521,7 @@ class PresencePushTestCase(unittest.TestCase):
         self.datastore.get_received_txn_response = get_received_txn_response
 
         self.handler = hs.get_handlers().presence_handler
-        self.handler.push_update_to_clients = self.mock_update_client
+        self.event_source = hs.get_event_sources().sources["presence"]
 
         # Mock the RoomMemberHandler
         hs.handlers.room_member_handler = Mock(spec=[
@@ -622,16 +615,23 @@ class PresencePushTestCase(unittest.TestCase):
         apple_set.add(self.u_banana)
         apple_set.add(self.u_clementine)
 
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
         yield self.handler.set_state(self.u_apple, self.u_apple,
                 {"state": ONLINE})
 
-        self.mock_update_client.assert_has_calls([
-                call(users_to_push=set([self.u_apple, self.u_banana, self.u_clementine]),
-                    room_ids=["a-room"],
-                    observed_user=self.u_apple,
-                    statuscache=ANY), # self-reflection
-        ], any_order=True)
-        self.mock_update_client.reset_mock()
+        self.assertEquals(self.event_source.get_current_key(), 1)
+        self.assertEquals(
+            self.event_source.get_new_events_for_user(self.u_apple, 0, None)[0],
+            [
+                {"type": "m.presence",
+                 "content": {
+                    "user_id": "@apple:test",
+                    "state": ONLINE,
+                    "mtime_age": 0,
+                }},
+            ],
+        )
 
         presence = yield self.handler.get_presence_list(
                 observer_user=self.u_apple, accepted=True)
@@ -657,31 +657,24 @@ class PresencePushTestCase(unittest.TestCase):
                  "state": OFFLINE},
         ], presence)
 
-        self.mock_update_client.assert_has_calls([
-                call(users_to_push=set([self.u_banana]),
-                    room_ids=[],
-                    observed_user=self.u_banana,
-                    statuscache=ANY), # self-reflection
-        ]) # and no others...
+        self.assertEquals(self.event_source.get_current_key(), 2)
+        self.assertEquals(
+            self.event_source.get_new_events_for_user(
+                self.u_banana, 1, None
+            )[0],
+            [
+                {"type": "m.presence",
+                 "content": {
+                     "user_id": "@banana:test",
+                     "state": ONLINE,
+                     "mtime_age": 2000
+                }},
+            ]
+        )
 
     @defer.inlineCallbacks
     def test_push_remote(self):
         put_json = self.mock_http_client.put_json
-#        put_json.expect_call_and_return(
-#            call("remote",
-#                path=ANY,  # Can't guarantee which txn ID will be which
-#                data=_expect_edu("remote", "m.presence",
-#                    content={
-#                        "push": [
-#                            {"user_id": "@apple:test",
-#                             "state": "online",
-#                             "mtime_age": 0},
-#                        ],
-#                    }
-#                )
-#            ),
-#            defer.succeed((200, "OK"))
-#        )
         put_json.expect_call_and_return(
             call("farm",
                 path=ANY,  # Can't guarantee which txn ID will be which
@@ -724,6 +717,8 @@ class PresencePushTestCase(unittest.TestCase):
 
         self.room_members = [self.u_banana, self.u_potato]
 
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
         yield self.mock_federation_resource.trigger("PUT",
             "/matrix/federation/v1/send/1000000/",
             _make_edu_json("elsewhere", "m.presence",
@@ -737,12 +732,20 @@ class PresencePushTestCase(unittest.TestCase):
             )
         )
 
-        self.mock_update_client.assert_has_calls([
-                call(users_to_push=set([self.u_apple]),
-                    room_ids=["a-room"],
-                    observed_user=self.u_potato,
-                    statuscache=ANY),
-        ], any_order=True)
+        self.assertEquals(self.event_source.get_current_key(), 1)
+        self.assertEquals(
+            self.event_source.get_new_events_for_user(
+                self.u_apple, 0, None
+            )[0],
+            [
+                {"type": "m.presence",
+                 "content": {
+                     "user_id": "@potato:remote",
+                     "state": ONLINE,
+                     "mtime_age": 1000,
+                }}
+            ]
+        )
 
         self.clock.advance_time(2)
 
@@ -754,24 +757,35 @@ class PresencePushTestCase(unittest.TestCase):
     def test_join_room_local(self):
         self.room_members = [self.u_apple, self.u_banana]
 
-        yield self.distributor.fire("user_joined_room", self.u_elderberry,
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
+        # TODO(paul): Gut-wrenching
+        self.handler._user_cachemap[self.u_clementine] = UserPresenceCache()
+        self.handler._user_cachemap[self.u_clementine].update(
+            {
+                "state": PresenceState.ONLINE,
+                "mtime": self.clock.time_msec(),
+            }, self.u_clementine
+        )
+
+        yield self.distributor.fire("user_joined_room", self.u_clementine,
             "a-room"
         )
 
-        self.mock_update_client.assert_has_calls([
-            call(room_ids=["a-room"],
-                observed_user=self.u_elderberry,
-                users_to_push=set(),
-                statuscache=ANY),
-            call(users_to_push=set([self.u_elderberry]),
-                observed_user=self.u_apple,
-                room_ids=[],
-                statuscache=ANY),
-            call(users_to_push=set([self.u_elderberry]),
-                observed_user=self.u_banana,
-                room_ids=[],
-                statuscache=ANY),
-        ], any_order=True)
+        self.assertEquals(self.event_source.get_current_key(), 1)
+        self.assertEquals(
+            self.event_source.get_new_events_for_user(
+                self.u_apple, 0, None
+            )[0],
+            [
+                {"type": "m.presence",
+                 "content": {
+                     "user_id": "@clementine:test",
+                     "state": ONLINE,
+                     "mtime_age": 0,
+                }}
+            ]
+        )
 
     @defer.inlineCallbacks
     def test_join_room_remote(self):
-- 
cgit 1.4.1