Hook up the event stream to typing notifications

2 files changed, 82 insertions, 2 deletions

diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 2370ff7134..7426bda960 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -183,12 +183,32 @@ class TypingNotificationHandler(BaseHandler):
 class TypingNotificationEventSource(object):
     def __init__(self, hs):
         self.hs = hs
+        self.handler = hs.get_handlers().typing_notification_handler
+
+    def _make_event_for(self, room_id):
+        typing = self.handler._room_typing[room_id]
+        return {
+            "type": "m.typing",
+            "room_id": room_id,
+            "typing": [u.to_string() for u in typing],
+        }
 
     def get_new_events_for_user(self, user, from_key, limit):
-        return ([], from_key)
+        from_key = int(from_key)
+        handler = self.handler
+
+        events = []
+        for room_id in handler._room_serials:
+            if handler._room_serials[room_id] <= from_key:
+                continue
+
+            # TODO: check if user is in room
+            events.append(self._make_event_for(room_id))
+
+        return (events, handler._latest_room_serial)
 
     def get_current_key(self):
-        return 0
+        return self.handler._latest_room_serial
 
     def get_pagination_rows(self, user, pagination_config, key):
         return ([], pagination_config.from_key)
diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py
index 898977ed8d..b858f96323 100644
--- a/tests/handlers/test_typing.py
+++ b/tests/handlers/test_typing.py
@@ -91,6 +91,8 @@ class TypingNotificationsTestCase(unittest.TestCase):
 
         self.handler = hs.get_handlers().typing_notification_handler
 
+        self.event_source = hs.get_event_sources().sources["typing"]
+
         self.datastore = hs.get_datastore()
         self.datastore.get_destination_retry_timings.return_value = (
             defer.succeed(DestinationsTable.EntryType("", 0, 0))
@@ -151,6 +153,8 @@ class TypingNotificationsTestCase(unittest.TestCase):
     def test_started_typing_local(self):
         self.room_members = [self.u_apple, self.u_banana]
 
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
         yield self.handler.started_typing(
             target_user=self.u_apple,
             auth_user=self.u_apple,
@@ -162,6 +166,16 @@ class TypingNotificationsTestCase(unittest.TestCase):
             call(rooms=[self.room_id]),
         ])
 
+        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.typing",
+                 "room_id": self.room_id,
+                 "typing": [self.u_apple.to_string()]},
+            ]
+        )
+
     @defer.inlineCallbacks
     def test_started_typing_remote_send(self):
         self.room_members = [self.u_apple, self.u_onion]
@@ -195,6 +209,8 @@ class TypingNotificationsTestCase(unittest.TestCase):
     def test_started_typing_remote_recv(self):
         self.room_members = [self.u_apple, self.u_onion]
 
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
         yield self.mock_federation_resource.trigger("PUT",
             "/_matrix/federation/v1/send/1000000/",
             _make_edu_json("farm", "m.typing",
@@ -210,6 +226,16 @@ class TypingNotificationsTestCase(unittest.TestCase):
             call(rooms=[self.room_id]),
         ])
 
+        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.typing",
+                 "room_id": self.room_id,
+                 "typing": [self.u_onion.to_string()]},
+            ]
+        )
+
     @defer.inlineCallbacks
     def test_stopped_typing(self):
         self.room_members = [self.u_apple, self.u_banana, self.u_onion]
@@ -239,6 +265,8 @@ class TypingNotificationsTestCase(unittest.TestCase):
         )
         self.handler._room_typing[self.room_id] = set((self.u_apple,))
 
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
         yield self.handler.stopped_typing(
             target_user=self.u_apple,
             auth_user=self.u_apple,
@@ -251,10 +279,22 @@ class TypingNotificationsTestCase(unittest.TestCase):
 
         yield put_json.await_calls()
 
+        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.typing",
+                 "room_id": self.room_id,
+                 "typing": []},
+            ]
+        )
+
     @defer.inlineCallbacks
     def test_typing_timeout(self):
         self.room_members = [self.u_apple, self.u_banana]
 
+        self.assertEquals(self.event_source.get_current_key(), 0)
+
         yield self.handler.started_typing(
             target_user=self.u_apple,
             auth_user=self.u_apple,
@@ -267,8 +307,28 @@ class TypingNotificationsTestCase(unittest.TestCase):
         ])
         self.on_new_user_event.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.typing",
+                 "room_id": self.room_id,
+                 "typing": [self.u_apple.to_string()]},
+            ]
+        )
+
         self.clock.advance_time(11)
 
         self.on_new_user_event.assert_has_calls([
             call(rooms=[self.room_id]),
         ])
+
+        self.assertEquals(self.event_source.get_current_key(), 2)
+        self.assertEquals(
+            self.event_source.get_new_events_for_user(self.u_apple, 1, None)[0],
+            [
+                {"type": "m.typing",
+                 "room_id": self.room_id,
+                 "typing": []},
+            ]
+        )