mirror of
https://mau.dev/maunium/synapse.git
synced 2024-10-01 01:36:05 -04:00
60a0f81c7a
synapse This is necessary for replicating the data in synapse to be visible to a separate service because presence and typing notifications aren't stored in a database so won't be visible to another process. This API can be used to either get the raw data by requesting the tables themselves or to just receive notifications for updates by following the streams meta-stream. Returns updates for each table requested a JSON array of arrays with a row for each row in the table. Each table is prefixed by a header row with the: name of the table, current stream_id position for the table, number of rows, number of columns and the names of the columns. This is followed by the rows that have been added to the server since the requester last asked. The API has a timeout and is hooked up to the notifier so that a slave can long poll for updates.
286 lines
9.1 KiB
Python
286 lines
9.1 KiB
Python
# -*- 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)
|