Convert user directory handler and related classes to async/await. (#7640)

1 files changed, 46 insertions, 72 deletions

diff --git a/synapse/handlers/user_directory.py b/synapse/handlers/user_directory.py
index 722760c59d..12423b909a 100644
--- a/synapse/handlers/user_directory.py
+++ b/synapse/handlers/user_directory.py
@@ -17,14 +17,11 @@ import logging
 
 from six import iteritems, iterkeys
 
-from twisted.internet import defer
-
 import synapse.metrics
 from synapse.api.constants import EventTypes, JoinRules, Membership
 from synapse.handlers.state_deltas import StateDeltasHandler
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.roommember import ProfileInfo
-from synapse.types import get_localpart_from_id
 from synapse.util.metrics import Measure
 
 logger = logging.getLogger(__name__)
@@ -103,43 +100,39 @@ class UserDirectoryHandler(StateDeltasHandler):
         if self._is_processing:
             return
 
-        @defer.inlineCallbacks
-        def process():
+        async def process():
             try:
-                yield self._unsafe_process()
+                await self._unsafe_process()
             finally:
                 self._is_processing = False
 
         self._is_processing = True
         run_as_background_process("user_directory.notify_new_event", process)
 
-    @defer.inlineCallbacks
-    def handle_local_profile_change(self, user_id, profile):
+    async def handle_local_profile_change(self, user_id, profile):
         """Called to update index of our local user profiles when they change
         irrespective of any rooms the user may be in.
         """
         # FIXME(#3714): We should probably do this in the same worker as all
         # the other changes.
-        is_support = yield self.store.is_support_user(user_id)
+        is_support = await self.store.is_support_user(user_id)
         # Support users are for diagnostics and should not appear in the user directory.
         if not is_support:
-            yield self.store.update_profile_in_user_dir(
+            await self.store.update_profile_in_user_dir(
                 user_id, profile.display_name, profile.avatar_url
             )
 
-    @defer.inlineCallbacks
-    def handle_user_deactivated(self, user_id):
+    async def handle_user_deactivated(self, user_id):
         """Called when a user ID is deactivated
         """
         # FIXME(#3714): We should probably do this in the same worker as all
         # the other changes.
-        yield self.store.remove_from_user_dir(user_id)
+        await self.store.remove_from_user_dir(user_id)
 
-    @defer.inlineCallbacks
-    def _unsafe_process(self):
+    async def _unsafe_process(self):
         # If self.pos is None then means we haven't fetched it from DB
         if self.pos is None:
-            self.pos = yield self.store.get_user_directory_stream_pos()
+            self.pos = await self.store.get_user_directory_stream_pos()
 
         # If still None then the initial background update hasn't happened yet
         if self.pos is None:
@@ -155,12 +148,12 @@ class UserDirectoryHandler(StateDeltasHandler):
                 logger.debug(
                     "Processing user stats %s->%s", self.pos, room_max_stream_ordering
                 )
-                max_pos, deltas = yield self.store.get_current_state_deltas(
+                max_pos, deltas = await self.store.get_current_state_deltas(
                     self.pos, room_max_stream_ordering
                 )
 
                 logger.debug("Handling %d state deltas", len(deltas))
-                yield self._handle_deltas(deltas)
+                await self._handle_deltas(deltas)
 
                 self.pos = max_pos
 
@@ -169,10 +162,9 @@ class UserDirectoryHandler(StateDeltasHandler):
                     max_pos
                 )
 
-                yield self.store.update_user_directory_stream_pos(max_pos)
+                await self.store.update_user_directory_stream_pos(max_pos)
 
-    @defer.inlineCallbacks
-    def _handle_deltas(self, deltas):
+    async def _handle_deltas(self, deltas):
         """Called with the state deltas to process
         """
         for delta in deltas:
@@ -187,11 +179,11 @@ class UserDirectoryHandler(StateDeltasHandler):
             # For join rule and visibility changes we need to check if the room
             # may have become public or not and add/remove the users in said room
             if typ in (EventTypes.RoomHistoryVisibility, EventTypes.JoinRules):
-                yield self._handle_room_publicity_change(
+                await self._handle_room_publicity_change(
                     room_id, prev_event_id, event_id, typ
                 )
             elif typ == EventTypes.Member:
-                change = yield self._get_key_change(
+                change = await self._get_key_change(
                     prev_event_id,
                     event_id,
                     key_name="membership",
@@ -201,7 +193,7 @@ class UserDirectoryHandler(StateDeltasHandler):
                 if change is False:
                     # Need to check if the server left the room entirely, if so
                     # we might need to remove all the users in that room
-                    is_in_room = yield self.store.is_host_joined(
+                    is_in_room = await self.store.is_host_joined(
                         room_id, self.server_name
                     )
                     if not is_in_room:
@@ -209,40 +201,41 @@ class UserDirectoryHandler(StateDeltasHandler):
                         # Fetch all the users that we marked as being in user
                         # directory due to being in the room and then check if
                         # need to remove those users or not
-                        user_ids = yield self.store.get_users_in_dir_due_to_room(
+                        user_ids = await self.store.get_users_in_dir_due_to_room(
                             room_id
                         )
 
                         for user_id in user_ids:
-                            yield self._handle_remove_user(room_id, user_id)
+                            await self._handle_remove_user(room_id, user_id)
                         return
                     else:
                         logger.debug("Server is still in room: %r", room_id)
 
-                is_support = yield self.store.is_support_user(state_key)
+                is_support = await self.store.is_support_user(state_key)
                 if not is_support:
                     if change is None:
                         # Handle any profile changes
-                        yield self._handle_profile_change(
+                        await self._handle_profile_change(
                             state_key, room_id, prev_event_id, event_id
                         )
                         continue
 
                     if change:  # The user joined
-                        event = yield self.store.get_event(event_id, allow_none=True)
+                        event = await self.store.get_event(event_id, allow_none=True)
                         profile = ProfileInfo(
                             avatar_url=event.content.get("avatar_url"),
                             display_name=event.content.get("displayname"),
                         )
 
-                        yield self._handle_new_user(room_id, state_key, profile)
+                        await self._handle_new_user(room_id, state_key, profile)
                     else:  # The user left
-                        yield self._handle_remove_user(room_id, state_key)
+                        await self._handle_remove_user(room_id, state_key)
             else:
                 logger.debug("Ignoring irrelevant type: %r", typ)
 
-    @defer.inlineCallbacks
-    def _handle_room_publicity_change(self, room_id, prev_event_id, event_id, typ):
+    async def _handle_room_publicity_change(
+        self, room_id, prev_event_id, event_id, typ
+    ):
         """Handle a room having potentially changed from/to world_readable/publically
         joinable.
 
@@ -255,14 +248,14 @@ class UserDirectoryHandler(StateDeltasHandler):
         logger.debug("Handling change for %s: %s", typ, room_id)
 
         if typ == EventTypes.RoomHistoryVisibility:
-            change = yield self._get_key_change(
+            change = await self._get_key_change(
                 prev_event_id,
                 event_id,
                 key_name="history_visibility",
                 public_value="world_readable",
             )
         elif typ == EventTypes.JoinRules:
-            change = yield self._get_key_change(
+            change = await self._get_key_change(
                 prev_event_id,
                 event_id,
                 key_name="join_rule",
@@ -278,7 +271,7 @@ class UserDirectoryHandler(StateDeltasHandler):
 
         # There's been a change to or from being world readable.
 
-        is_public = yield self.store.is_room_world_readable_or_publicly_joinable(
+        is_public = await self.store.is_room_world_readable_or_publicly_joinable(
             room_id
         )
 
@@ -293,11 +286,11 @@ class UserDirectoryHandler(StateDeltasHandler):
             # ignore the change
             return
 
-        users_with_profile = yield self.state.get_current_users_in_room(room_id)
+        users_with_profile = await self.state.get_current_users_in_room(room_id)
 
         # Remove every user from the sharing tables for that room.
         for user_id in iterkeys(users_with_profile):
-            yield self.store.remove_user_who_share_room(user_id, room_id)
+            await self.store.remove_user_who_share_room(user_id, room_id)
 
         # Then, re-add them to the tables.
         # NOTE: this is not the most efficient method, as handle_new_user sets
@@ -306,26 +299,9 @@ class UserDirectoryHandler(StateDeltasHandler):
         # being added multiple times. The batching upserts shouldn't make this
         # too bad, though.
         for user_id, profile in iteritems(users_with_profile):
-            yield self._handle_new_user(room_id, user_id, profile)
-
-    @defer.inlineCallbacks
-    def _handle_local_user(self, user_id):
-        """Adds a new local roomless user into the user_directory_search table.
-        Used to populate up the user index when we have an
-        user_directory_search_all_users specified.
-        """
-        logger.debug("Adding new local user to dir, %r", user_id)
-
-        profile = yield self.store.get_profileinfo(get_localpart_from_id(user_id))
-
-        row = yield self.store.get_user_in_directory(user_id)
-        if not row:
-            yield self.store.update_profile_in_user_dir(
-                user_id, profile.display_name, profile.avatar_url
-            )
+            await self._handle_new_user(room_id, user_id, profile)
 
-    @defer.inlineCallbacks
-    def _handle_new_user(self, room_id, user_id, profile):
+    async def _handle_new_user(self, room_id, user_id, profile):
         """Called when we might need to add user to directory
 
         Args:
@@ -334,18 +310,18 @@ class UserDirectoryHandler(StateDeltasHandler):
         """
         logger.debug("Adding new user to dir, %r", user_id)
 
-        yield self.store.update_profile_in_user_dir(
+        await self.store.update_profile_in_user_dir(
             user_id, profile.display_name, profile.avatar_url
         )
 
-        is_public = yield self.store.is_room_world_readable_or_publicly_joinable(
+        is_public = await self.store.is_room_world_readable_or_publicly_joinable(
             room_id
         )
         # Now we update users who share rooms with users.
-        users_with_profile = yield self.state.get_current_users_in_room(room_id)
+        users_with_profile = await self.state.get_current_users_in_room(room_id)
 
         if is_public:
-            yield self.store.add_users_in_public_rooms(room_id, (user_id,))
+            await self.store.add_users_in_public_rooms(room_id, (user_id,))
         else:
             to_insert = set()
 
@@ -376,10 +352,9 @@ class UserDirectoryHandler(StateDeltasHandler):
                     to_insert.add((other_user_id, user_id))
 
             if to_insert:
-                yield self.store.add_users_who_share_private_room(room_id, to_insert)
+                await self.store.add_users_who_share_private_room(room_id, to_insert)
 
-    @defer.inlineCallbacks
-    def _handle_remove_user(self, room_id, user_id):
+    async def _handle_remove_user(self, room_id, user_id):
         """Called when we might need to remove user from directory
 
         Args:
@@ -389,24 +364,23 @@ class UserDirectoryHandler(StateDeltasHandler):
         logger.debug("Removing user %r", user_id)
 
         # Remove user from sharing tables
-        yield self.store.remove_user_who_share_room(user_id, room_id)
+        await self.store.remove_user_who_share_room(user_id, room_id)
 
         # Are they still in any rooms? If not, remove them entirely.
-        rooms_user_is_in = yield self.store.get_user_dir_rooms_user_is_in(user_id)
+        rooms_user_is_in = await self.store.get_user_dir_rooms_user_is_in(user_id)
 
         if len(rooms_user_is_in) == 0:
-            yield self.store.remove_from_user_dir(user_id)
+            await self.store.remove_from_user_dir(user_id)
 
-    @defer.inlineCallbacks
-    def _handle_profile_change(self, user_id, room_id, prev_event_id, event_id):
+    async def _handle_profile_change(self, user_id, room_id, prev_event_id, event_id):
         """Check member event changes for any profile changes and update the
         database if there are.
         """
         if not prev_event_id or not event_id:
             return
 
-        prev_event = yield self.store.get_event(prev_event_id, allow_none=True)
-        event = yield self.store.get_event(event_id, allow_none=True)
+        prev_event = await self.store.get_event(prev_event_id, allow_none=True)
+        event = await self.store.get_event(event_id, allow_none=True)
 
         if not prev_event or not event:
             return
@@ -421,4 +395,4 @@ class UserDirectoryHandler(StateDeltasHandler):
         new_avatar = event.content.get("avatar_url")
 
         if prev_name != new_name or prev_avatar != new_avatar:
-            yield self.store.update_profile_in_user_dir(user_id, new_name, new_avatar)
+            await self.store.update_profile_in_user_dir(user_id, new_name, new_avatar)