Merge branch 'release-v0.16.0' of github.com:matrix-org/synapse v0.16.0

1 files changed, 21 insertions, 23 deletions

diff --git a/synapse/notifier.py b/synapse/notifier.py
index f00cd8c588..30883a0696 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -14,13 +14,14 @@
 # limitations under the License.
 
 from twisted.internet import defer
-from synapse.api.constants import EventTypes
+from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import AuthError
 
 from synapse.util.logutils import log_function
 from synapse.util.async import ObservableDeferred
 from synapse.util.logcontext import PreserveLoggingContext
 from synapse.types import StreamToken
+from synapse.visibility import filter_events_for_client
 import synapse.metrics
 
 from collections import namedtuple
@@ -139,8 +140,6 @@ class Notifier(object):
     UNUSED_STREAM_EXPIRY_MS = 10 * 60 * 1000
 
     def __init__(self, hs):
-        self.hs = hs
-
         self.user_to_user_stream = {}
         self.room_to_user_streams = {}
         self.appservice_to_user_streams = {}
@@ -150,10 +149,8 @@ class Notifier(object):
         self.pending_new_room_events = []
 
         self.clock = hs.get_clock()
-
-        hs.get_distributor().observe(
-            "user_joined_room", self._user_joined_room
-        )
+        self.appservice_handler = hs.get_application_service_handler()
+        self.state_handler = hs.get_state_handler()
 
         self.clock.looping_call(
             self.remove_expired_streams, self.UNUSED_STREAM_EXPIRY_MS
@@ -231,9 +228,7 @@ class Notifier(object):
     def _on_new_room_event(self, event, room_stream_id, extra_users=[]):
         """Notify any user streams that are interested in this room event"""
         # poke any interested application service.
-        self.hs.get_handlers().appservice_handler.notify_interested_services(
-            event
-        )
+        self.appservice_handler.notify_interested_services(event)
 
         app_streams = set()
 
@@ -249,6 +244,9 @@ class Notifier(object):
                 )
                 app_streams |= app_user_streams
 
+        if event.type == EventTypes.Member and event.membership == Membership.JOIN:
+            self._user_joined_room(event.state_key, event.room_id)
+
         self.on_new_event(
             "room_key", room_stream_id,
             users=extra_users,
@@ -398,8 +396,8 @@ class Notifier(object):
                 )
 
                 if name == "room":
-                    room_member_handler = self.hs.get_handlers().room_member_handler
-                    new_events = yield room_member_handler._filter_events_for_client(
+                    new_events = yield filter_events_for_client(
+                        self.store,
                         user.to_string(),
                         new_events,
                         is_peeking=is_peeking,
@@ -448,7 +446,7 @@ class Notifier(object):
 
     @defer.inlineCallbacks
     def _is_world_readable(self, room_id):
-        state = yield self.hs.get_state_handler().get_current_state(
+        state = yield self.state_handler.get_current_state(
             room_id,
             EventTypes.RoomHistoryVisibility
         )
@@ -484,9 +482,8 @@ class Notifier(object):
                 user_stream.appservice, set()
             ).add(user_stream)
 
-    def _user_joined_room(self, user, room_id):
-        user = str(user)
-        new_user_stream = self.user_to_user_stream.get(user)
+    def _user_joined_room(self, user_id, room_id):
+        new_user_stream = self.user_to_user_stream.get(user_id)
         if new_user_stream is not None:
             room_streams = self.room_to_user_streams.setdefault(room_id, set())
             room_streams.add(new_user_stream)
@@ -503,13 +500,14 @@ class Notifier(object):
     def wait_for_replication(self, callback, timeout):
         """Wait for an event to happen.
 
-        :param callback:
-            Gets called whenever an event happens. If this returns a truthy
-            value then ``wait_for_replication`` returns, otherwise it waits
-            for another event.
-        :param int timeout:
-            How many milliseconds to wait for callback return a truthy value.
-        :returns:
+        Args:
+            callback: Gets called whenever an event happens. If this returns a
+                truthy value then ``wait_for_replication`` returns, otherwise
+                it waits for another event.
+            timeout: How many milliseconds to wait for callback return a truthy
+                value.
+
+        Returns:
             A deferred that resolves with the value returned by the callback.
         """
         listener = _NotificationListener(None)