mirror of
https://git.anonymousland.org/anonymousland/synapse.git
synced 2025-08-03 19:44:14 -04:00
Merge branch 'develop' of github.com:matrix-org/synapse into matrix-org-hotfixes
This commit is contained in:
commit
b6b1382be1
14 changed files with 152 additions and 59 deletions
|
@ -19,6 +19,7 @@ from synapse.api.constants import EventTypes, Membership
|
|||
from synapse.api.errors import AuthError, Codes
|
||||
from synapse.events.utils import serialize_event
|
||||
from synapse.events.validator import EventValidator
|
||||
from synapse.handlers.presence import format_user_presence_state
|
||||
from synapse.streams.config import PaginationConfig
|
||||
from synapse.types import (
|
||||
UserID, StreamToken,
|
||||
|
@ -225,9 +226,17 @@ class InitialSyncHandler(BaseHandler):
|
|||
"content": content,
|
||||
})
|
||||
|
||||
now = self.clock.time_msec()
|
||||
|
||||
ret = {
|
||||
"rooms": rooms_ret,
|
||||
"presence": presence,
|
||||
"presence": [
|
||||
{
|
||||
"type": "m.presence",
|
||||
"content": format_user_presence_state(event, now),
|
||||
}
|
||||
for event in presence
|
||||
],
|
||||
"account_data": account_data_events,
|
||||
"receipts": receipt,
|
||||
"end": now_token.to_string(),
|
||||
|
|
|
@ -29,6 +29,7 @@ from synapse.api.errors import SynapseError
|
|||
from synapse.api.constants import PresenceState
|
||||
from synapse.storage.presence import UserPresenceState
|
||||
|
||||
from synapse.util.caches.descriptors import cachedInlineCallbacks
|
||||
from synapse.util.logcontext import preserve_fn
|
||||
from synapse.util.logutils import log_function
|
||||
from synapse.util.metrics import Measure
|
||||
|
@ -722,9 +723,7 @@ class PresenceHandler(object):
|
|||
for state in updates
|
||||
])
|
||||
else:
|
||||
defer.returnValue([
|
||||
format_user_presence_state(state, now) for state in updates
|
||||
])
|
||||
defer.returnValue(updates)
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def set_state(self, target_user, state, ignore_status_msg=False):
|
||||
|
@ -798,6 +797,9 @@ class PresenceHandler(object):
|
|||
as_event=False,
|
||||
)
|
||||
|
||||
now = self.clock.time_msec()
|
||||
results[:] = [format_user_presence_state(r, now) for r in results]
|
||||
|
||||
is_accepted = {
|
||||
row["observed_user_id"]: row["accepted"] for row in presence_list
|
||||
}
|
||||
|
@ -850,6 +852,7 @@ class PresenceHandler(object):
|
|||
)
|
||||
|
||||
state_dict = yield self.get_state(observed_user, as_event=False)
|
||||
state_dict = format_user_presence_state(state_dict, self.clock.time_msec())
|
||||
|
||||
self.federation.send_edu(
|
||||
destination=observer_user.domain,
|
||||
|
@ -982,14 +985,18 @@ def should_notify(old_state, new_state):
|
|||
return False
|
||||
|
||||
|
||||
def format_user_presence_state(state, now):
|
||||
def format_user_presence_state(state, now, include_user_id=True):
|
||||
"""Convert UserPresenceState to a format that can be sent down to clients
|
||||
and to other servers.
|
||||
|
||||
The "user_id" is optional so that this function can be used to format presence
|
||||
updates for client /sync responses and for federation /send requests.
|
||||
"""
|
||||
content = {
|
||||
"presence": state.state,
|
||||
"user_id": state.user_id,
|
||||
}
|
||||
if include_user_id:
|
||||
content["user_id"] = state.user_id
|
||||
if state.last_active_ts:
|
||||
content["last_active_ago"] = now - state.last_active_ts
|
||||
if state.status_msg and state.state != PresenceState.OFFLINE:
|
||||
|
@ -1028,7 +1035,6 @@ class PresenceEventSource(object):
|
|||
# sending down the rare duplicate is not a concern.
|
||||
|
||||
with Measure(self.clock, "presence.get_new_events"):
|
||||
user_id = user.to_string()
|
||||
if from_key is not None:
|
||||
from_key = int(from_key)
|
||||
|
||||
|
@ -1037,18 +1043,7 @@ class PresenceEventSource(object):
|
|||
|
||||
max_token = self.store.get_current_presence_token()
|
||||
|
||||
plist = yield self.store.get_presence_list_accepted(user.localpart)
|
||||
users_interested_in = set(row["observed_user_id"] for row in plist)
|
||||
users_interested_in.add(user_id) # So that we receive our own presence
|
||||
|
||||
users_who_share_room = yield self.store.get_users_who_share_room_with_user(
|
||||
user_id
|
||||
)
|
||||
users_interested_in.update(users_who_share_room)
|
||||
|
||||
if explicit_room_id:
|
||||
user_ids = yield self.store.get_users_in_room(explicit_room_id)
|
||||
users_interested_in.update(user_ids)
|
||||
users_interested_in = yield self._get_interested_in(user, explicit_room_id)
|
||||
|
||||
user_ids_changed = set()
|
||||
changed = None
|
||||
|
@ -1076,16 +1071,13 @@ class PresenceEventSource(object):
|
|||
|
||||
updates = yield presence.current_state_for_users(user_ids_changed)
|
||||
|
||||
now = self.clock.time_msec()
|
||||
|
||||
defer.returnValue(([
|
||||
{
|
||||
"type": "m.presence",
|
||||
"content": format_user_presence_state(s, now),
|
||||
}
|
||||
for s in updates.values()
|
||||
if include_offline or s.state != PresenceState.OFFLINE
|
||||
], max_token))
|
||||
if include_offline:
|
||||
defer.returnValue((updates.values(), max_token))
|
||||
else:
|
||||
defer.returnValue(([
|
||||
s for s in updates.itervalues()
|
||||
if s.state != PresenceState.OFFLINE
|
||||
], max_token))
|
||||
|
||||
def get_current_key(self):
|
||||
return self.store.get_current_presence_token()
|
||||
|
@ -1093,6 +1085,31 @@ class PresenceEventSource(object):
|
|||
def get_pagination_rows(self, user, pagination_config, key):
|
||||
return self.get_new_events(user, from_key=None, include_offline=False)
|
||||
|
||||
@cachedInlineCallbacks(num_args=2, cache_context=True)
|
||||
def _get_interested_in(self, user, explicit_room_id, cache_context):
|
||||
"""Returns the set of users that the given user should see presence
|
||||
updates for
|
||||
"""
|
||||
user_id = user.to_string()
|
||||
plist = yield self.store.get_presence_list_accepted(
|
||||
user.localpart, on_invalidate=cache_context.invalidate,
|
||||
)
|
||||
users_interested_in = set(row["observed_user_id"] for row in plist)
|
||||
users_interested_in.add(user_id) # So that we receive our own presence
|
||||
|
||||
users_who_share_room = yield self.store.get_users_who_share_room_with_user(
|
||||
user_id, on_invalidate=cache_context.invalidate,
|
||||
)
|
||||
users_interested_in.update(users_who_share_room)
|
||||
|
||||
if explicit_room_id:
|
||||
user_ids = yield self.store.get_users_in_room(
|
||||
explicit_room_id, on_invalidate=cache_context.invalidate,
|
||||
)
|
||||
users_interested_in.update(user_ids)
|
||||
|
||||
defer.returnValue(users_interested_in)
|
||||
|
||||
|
||||
def handle_timeouts(user_states, is_mine_fn, syncing_user_ids, now):
|
||||
"""Checks the presence of users that have timed out and updates as
|
||||
|
|
|
@ -721,14 +721,14 @@ class SyncHandler(object):
|
|||
extra_users_ids.update(users)
|
||||
extra_users_ids.discard(user.to_string())
|
||||
|
||||
states = yield self.presence_handler.get_states(
|
||||
extra_users_ids,
|
||||
as_event=True,
|
||||
)
|
||||
presence.extend(states)
|
||||
if extra_users_ids:
|
||||
states = yield self.presence_handler.get_states(
|
||||
extra_users_ids,
|
||||
)
|
||||
presence.extend(states)
|
||||
|
||||
# Deduplicate the presence entries so that there's at most one per user
|
||||
presence = {p["content"]["user_id"]: p for p in presence}.values()
|
||||
# Deduplicate the presence entries so that there's at most one per user
|
||||
presence = {p.user_id: p for p in presence}.values()
|
||||
|
||||
presence = sync_config.filter_collection.filter_presence(
|
||||
presence
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue