123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285 |
- # -*- coding: utf-8 -*-
- # Copyright 2014-2016 OpenMarket Ltd
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- from twisted.internet import defer
- from ._base import BaseHandler
- from synapse.api.errors import SynapseError, AuthError
- from synapse.util.logcontext import PreserveLoggingContext
- from synapse.util.metrics import Measure
- from synapse.types import UserID
- import logging
- from collections import namedtuple
- import ujson as json
- logger = logging.getLogger(__name__)
- # A tiny object useful for storing a user's membership in a room, as a mapping
- # key
- RoomMember = namedtuple("RoomMember", ("room_id", "user"))
- class TypingNotificationHandler(BaseHandler):
- def __init__(self, hs):
- super(TypingNotificationHandler, self).__init__(hs)
- self.homeserver = hs
- self.clock = hs.get_clock()
- self.federation = hs.get_replication_layer()
- self.federation.register_edu_handler("m.typing", self._recv_edu)
- hs.get_distributor().observe("user_left_room", self.user_left_room)
- self._member_typing_until = {} # clock time we expect to stop
- self._member_typing_timer = {} # deferreds to manage theabove
- # map room IDs to serial numbers
- self._room_serials = {}
- self._latest_room_serial = 0
- # map room IDs to sets of users currently typing
- self._room_typing = {}
- def tearDown(self):
- """Cancels all the pending timers.
- Normally this shouldn't be needed, but it's required from unit tests
- to avoid a "Reactor was unclean" warning."""
- for t in self._member_typing_timer.values():
- self.clock.cancel_call_later(t)
- @defer.inlineCallbacks
- def started_typing(self, target_user, auth_user, room_id, timeout):
- if not self.hs.is_mine(target_user):
- raise SynapseError(400, "User is not hosted on this Home Server")
- if target_user != auth_user:
- raise AuthError(400, "Cannot set another user's typing state")
- yield self.auth.check_joined_room(room_id, target_user.to_string())
- logger.debug(
- "%s has started typing in %s", target_user.to_string(), room_id
- )
- until = self.clock.time_msec() + timeout
- member = RoomMember(room_id=room_id, user=target_user)
- was_present = member in self._member_typing_until
- if member in self._member_typing_timer:
- self.clock.cancel_call_later(self._member_typing_timer[member])
- def _cb():
- logger.debug(
- "%s has timed out in %s", target_user.to_string(), room_id
- )
- self._stopped_typing(member)
- self._member_typing_until[member] = until
- self._member_typing_timer[member] = self.clock.call_later(
- timeout / 1000.0, _cb
- )
- if was_present:
- # No point sending another notification
- defer.returnValue(None)
- yield self._push_update(
- room_id=room_id,
- user=target_user,
- typing=True,
- )
- @defer.inlineCallbacks
- def stopped_typing(self, target_user, auth_user, room_id):
- if not self.hs.is_mine(target_user):
- raise SynapseError(400, "User is not hosted on this Home Server")
- if target_user != auth_user:
- raise AuthError(400, "Cannot set another user's typing state")
- yield self.auth.check_joined_room(room_id, target_user.to_string())
- logger.debug(
- "%s has stopped typing in %s", target_user.to_string(), room_id
- )
- member = RoomMember(room_id=room_id, user=target_user)
- if member in self._member_typing_timer:
- self.clock.cancel_call_later(self._member_typing_timer[member])
- del self._member_typing_timer[member]
- yield self._stopped_typing(member)
- @defer.inlineCallbacks
- def user_left_room(self, user, room_id):
- if self.hs.is_mine(user):
- member = RoomMember(room_id=room_id, user=user)
- yield self._stopped_typing(member)
- @defer.inlineCallbacks
- def _stopped_typing(self, member):
- if member not in self._member_typing_until:
- # No point
- defer.returnValue(None)
- yield self._push_update(
- room_id=member.room_id,
- user=member.user,
- typing=False,
- )
- del self._member_typing_until[member]
- if member in self._member_typing_timer:
- # Don't cancel it - either it already expired, or the real
- # stopped_typing() will cancel it
- del self._member_typing_timer[member]
- @defer.inlineCallbacks
- def _push_update(self, room_id, user, typing):
- localusers = set()
- remotedomains = set()
- rm_handler = self.homeserver.get_handlers().room_member_handler
- yield rm_handler.fetch_room_distributions_into(
- room_id, localusers=localusers, remotedomains=remotedomains
- )
- if localusers:
- self._push_update_local(
- room_id=room_id,
- user=user,
- typing=typing
- )
- deferreds = []
- for domain in remotedomains:
- deferreds.append(self.federation.send_edu(
- destination=domain,
- edu_type="m.typing",
- content={
- "room_id": room_id,
- "user_id": user.to_string(),
- "typing": typing,
- },
- ))
- yield defer.DeferredList(deferreds, consumeErrors=True)
- @defer.inlineCallbacks
- def _recv_edu(self, origin, content):
- room_id = content["room_id"]
- user = UserID.from_string(content["user_id"])
- localusers = set()
- rm_handler = self.homeserver.get_handlers().room_member_handler
- yield rm_handler.fetch_room_distributions_into(
- room_id, localusers=localusers
- )
- if localusers:
- self._push_update_local(
- room_id=room_id,
- user=user,
- typing=content["typing"]
- )
- def _push_update_local(self, room_id, user, typing):
- room_set = self._room_typing.setdefault(room_id, set())
- if typing:
- room_set.add(user)
- else:
- room_set.discard(user)
- self._latest_room_serial += 1
- self._room_serials[room_id] = self._latest_room_serial
- with PreserveLoggingContext():
- self.notifier.on_new_event(
- "typing_key", self._latest_room_serial, rooms=[room_id]
- )
- def get_all_typing_updates(self, last_id, current_id):
- # TODO: Work out a way to do this without scanning the entire state.
- rows = []
- for room_id, serial in self._room_serials.items():
- if last_id < serial and serial <= current_id:
- typing = self._room_typing[room_id]
- typing_bytes = json.dumps([
- u.to_string() for u in typing
- ], ensure_ascii=False)
- rows.append((serial, room_id, typing_bytes))
- rows.sort()
- return rows
- class TypingNotificationEventSource(object):
- def __init__(self, hs):
- self.hs = hs
- self.clock = hs.get_clock()
- self._handler = None
- self._room_member_handler = None
- def handler(self):
- # Avoid cyclic dependency in handler setup
- if not self._handler:
- self._handler = self.hs.get_handlers().typing_notification_handler
- return self._handler
- def room_member_handler(self):
- if not self._room_member_handler:
- self._room_member_handler = self.hs.get_handlers().room_member_handler
- return self._room_member_handler
- def _make_event_for(self, room_id):
- typing = self.handler()._room_typing[room_id]
- return {
- "type": "m.typing",
- "room_id": room_id,
- "content": {
- "user_ids": [u.to_string() for u in typing],
- },
- }
- def get_new_events(self, from_key, room_ids, **kwargs):
- with Measure(self.clock, "typing.get_new_events"):
- from_key = int(from_key)
- handler = self.handler()
- events = []
- for room_id in room_ids:
- if room_id not in handler._room_serials:
- continue
- if handler._room_serials[room_id] <= from_key:
- continue
- events.append(self._make_event_for(room_id))
- return events, handler._latest_room_serial
- def get_current_key(self):
- return self.handler()._latest_room_serial
- def get_pagination_rows(self, user, pagination_config, key):
- return ([], pagination_config.from_key)
|