mirror of
https://github.com/element-hq/synapse
synced 2024-08-19 08:00:53 +00:00
Add comments
This commit is contained in:
parent
c974116f19
commit
fbfad76c03
4 changed files with 61 additions and 7 deletions
|
@ -192,6 +192,9 @@ class DeviceHandler(BaseHandler):
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def notify_device_update(self, user_id, device_ids):
|
def notify_device_update(self, user_id, device_ids):
|
||||||
|
"""Notify that a user's device(s) has changed. Pokes the notifier, and
|
||||||
|
remote servers if the user is local.
|
||||||
|
"""
|
||||||
rooms = yield self.store.get_rooms_for_user(user_id)
|
rooms = yield self.store.get_rooms_for_user(user_id)
|
||||||
room_ids = [r.room_id for r in rooms]
|
room_ids = [r.room_id for r in rooms]
|
||||||
|
|
||||||
|
@ -210,12 +213,16 @@ class DeviceHandler(BaseHandler):
|
||||||
"device_list_key", position, rooms=room_ids,
|
"device_list_key", position, rooms=room_ids,
|
||||||
)
|
)
|
||||||
|
|
||||||
logger.info("Sending device list update notif to: %r", hosts)
|
if hosts:
|
||||||
for host in hosts:
|
logger.info("Sending device list update notif to: %r", hosts)
|
||||||
self.federation_sender.send_device_messages(host)
|
for host in hosts:
|
||||||
|
self.federation_sender.send_device_messages(host)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def get_device_list_changes(self, user_id, room_ids, from_key):
|
def get_device_list_changes(self, user_id, room_ids, from_key):
|
||||||
|
"""For a user and their joined rooms, calculate which device updates
|
||||||
|
we need to return.
|
||||||
|
"""
|
||||||
room_ids = frozenset(room_ids)
|
room_ids = frozenset(room_ids)
|
||||||
|
|
||||||
user_ids_changed = set()
|
user_ids_changed = set()
|
||||||
|
@ -236,11 +243,14 @@ class DeviceHandler(BaseHandler):
|
||||||
|
|
||||||
if get_domain_from_id(user_id) != origin:
|
if get_domain_from_id(user_id) != origin:
|
||||||
# TODO: Raise?
|
# TODO: Raise?
|
||||||
|
logger.warning("Got device list update edu for %r from %r", user_id, origin)
|
||||||
return
|
return
|
||||||
|
|
||||||
logger.info("Got edu: %r", edu_content)
|
logger.info("Got edu: %r", edu_content)
|
||||||
|
|
||||||
with (yield self._remote_edue_linearizer.queue(user_id)):
|
with (yield self._remote_edue_linearizer.queue(user_id)):
|
||||||
|
# If the prev id matches whats in our cache table, then we don't need
|
||||||
|
# to resync the users device list, otherwise we do.
|
||||||
resync = True
|
resync = True
|
||||||
if len(prev_ids) == 1:
|
if len(prev_ids) == 1:
|
||||||
extremity = yield self.store.get_device_list_remote_extremity(user_id)
|
extremity = yield self.store.get_device_list_remote_extremity(user_id)
|
||||||
|
@ -249,6 +259,7 @@ class DeviceHandler(BaseHandler):
|
||||||
resync = False
|
resync = False
|
||||||
|
|
||||||
if resync:
|
if resync:
|
||||||
|
# Fetch all devices for the user.
|
||||||
result = yield self.federation.query_user_devices(origin, user_id)
|
result = yield self.federation.query_user_devices(origin, user_id)
|
||||||
stream_id = result["stream_id"]
|
stream_id = result["stream_id"]
|
||||||
devices = result["devices"]
|
devices = result["devices"]
|
||||||
|
@ -258,6 +269,8 @@ class DeviceHandler(BaseHandler):
|
||||||
device_ids = [device["device_id"] for device in devices]
|
device_ids = [device["device_id"] for device in devices]
|
||||||
yield self.notify_device_update(user_id, device_ids)
|
yield self.notify_device_update(user_id, device_ids)
|
||||||
else:
|
else:
|
||||||
|
# Simply update the single device, since we know that is the only
|
||||||
|
# change (becuase of the single prev_id matching the current cache)
|
||||||
content = dict(edu_content)
|
content = dict(edu_content)
|
||||||
for key in ("user_id", "device_id", "stream_id", "prev_ids"):
|
for key in ("user_id", "device_id", "stream_id", "prev_ids"):
|
||||||
content.pop(key, None)
|
content.pop(key, None)
|
||||||
|
|
|
@ -75,7 +75,7 @@ class E2eKeysHandler(object):
|
||||||
else:
|
else:
|
||||||
remote_queries[user_id] = device_ids
|
remote_queries[user_id] = device_ids
|
||||||
|
|
||||||
# do the queries
|
# Firt get local devices.
|
||||||
failures = {}
|
failures = {}
|
||||||
results = {}
|
results = {}
|
||||||
if local_query:
|
if local_query:
|
||||||
|
@ -84,6 +84,7 @@ class E2eKeysHandler(object):
|
||||||
if user_id in local_query:
|
if user_id in local_query:
|
||||||
results[user_id] = keys
|
results[user_id] = keys
|
||||||
|
|
||||||
|
# Now attempt to get any remote devices from our local cache.
|
||||||
remote_queries_not_in_cache = {}
|
remote_queries_not_in_cache = {}
|
||||||
if remote_queries:
|
if remote_queries:
|
||||||
query_list = []
|
query_list = []
|
||||||
|
@ -115,6 +116,7 @@ class E2eKeysHandler(object):
|
||||||
r = remote_queries_not_in_cache.setdefault(domain, {})
|
r = remote_queries_not_in_cache.setdefault(domain, {})
|
||||||
r[user_id] = remote_queries[user_id]
|
r[user_id] = remote_queries[user_id]
|
||||||
|
|
||||||
|
# Now fetch any devices that we don't have in our cache
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def do_remote_query(destination):
|
def do_remote_query(destination):
|
||||||
destination_query = remote_queries_not_in_cache[destination]
|
destination_query = remote_queries_not_in_cache[destination]
|
||||||
|
|
|
@ -139,6 +139,9 @@ class DeviceStore(SQLBaseStore):
|
||||||
defer.returnValue({d["device_id"]: d for d in devices})
|
defer.returnValue({d["device_id"]: d for d in devices})
|
||||||
|
|
||||||
def get_device_list_remote_extremity(self, user_id):
|
def get_device_list_remote_extremity(self, user_id):
|
||||||
|
"""Get the last stream_id we got for a user. May be None if we haven't
|
||||||
|
got any information for them.
|
||||||
|
"""
|
||||||
return self._simple_select_one_onecol(
|
return self._simple_select_one_onecol(
|
||||||
table="device_lists_remote_extremeties",
|
table="device_lists_remote_extremeties",
|
||||||
keyvalues={"user_id": user_id},
|
keyvalues={"user_id": user_id},
|
||||||
|
@ -149,6 +152,8 @@ class DeviceStore(SQLBaseStore):
|
||||||
|
|
||||||
def update_remote_device_list_cache_entry(self, user_id, device_id, content,
|
def update_remote_device_list_cache_entry(self, user_id, device_id, content,
|
||||||
stream_id):
|
stream_id):
|
||||||
|
"""Updates a single user's device in the cache.
|
||||||
|
"""
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
"update_remote_device_list_cache_entry",
|
"update_remote_device_list_cache_entry",
|
||||||
self._update_remote_device_list_cache_entry_txn,
|
self._update_remote_device_list_cache_entry_txn,
|
||||||
|
@ -181,6 +186,8 @@ class DeviceStore(SQLBaseStore):
|
||||||
)
|
)
|
||||||
|
|
||||||
def update_remote_device_list_cache(self, user_id, devices, stream_id):
|
def update_remote_device_list_cache(self, user_id, devices, stream_id):
|
||||||
|
"""Replace the cache of the remote user's devices.
|
||||||
|
"""
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
"update_remote_device_list_cache",
|
"update_remote_device_list_cache",
|
||||||
self._update_remote_device_list_cache_txn,
|
self._update_remote_device_list_cache_txn,
|
||||||
|
@ -222,6 +229,11 @@ class DeviceStore(SQLBaseStore):
|
||||||
)
|
)
|
||||||
|
|
||||||
def get_devices_by_remote(self, destination, from_stream_id):
|
def get_devices_by_remote(self, destination, from_stream_id):
|
||||||
|
"""Get stream of updates to send to remote servers
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
(now_stream_id, [ { updates }, .. ])
|
||||||
|
"""
|
||||||
now_stream_id = self._device_list_id_gen.get_current_token()
|
now_stream_id = self._device_list_id_gen.get_current_token()
|
||||||
|
|
||||||
has_changed = self._device_list_federation_stream_cache.has_entity_changed(
|
has_changed = self._device_list_federation_stream_cache.has_entity_changed(
|
||||||
|
@ -290,6 +302,17 @@ class DeviceStore(SQLBaseStore):
|
||||||
return (now_stream_id, results)
|
return (now_stream_id, results)
|
||||||
|
|
||||||
def get_user_devices_from_cache(self, query_list):
|
def get_user_devices_from_cache(self, query_list):
|
||||||
|
"""Get the devices (and keys if any) for remote users from the cache.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
query_list(list): List of (user_id, device_ids), if device_ids is
|
||||||
|
falsey then return all device ids for that user.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
(user_ids_not_in_cache, results_map), where user_ids_not_in_cache is
|
||||||
|
a set of user_ids and results_map is a mapping of
|
||||||
|
user_id -> device_id -> device_info
|
||||||
|
"""
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
"get_user_devices_from_cache", self._get_user_devices_from_cache_txn,
|
"get_user_devices_from_cache", self._get_user_devices_from_cache_txn,
|
||||||
query_list,
|
query_list,
|
||||||
|
@ -347,6 +370,11 @@ class DeviceStore(SQLBaseStore):
|
||||||
return user_ids_not_in_cache, results
|
return user_ids_not_in_cache, results
|
||||||
|
|
||||||
def get_devices_with_keys_by_user(self, user_id):
|
def get_devices_with_keys_by_user(self, user_id):
|
||||||
|
"""Get all devices (with any device keys) for a user
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
(stream_id, devices)
|
||||||
|
"""
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
"get_devices_with_keys_by_user",
|
"get_devices_with_keys_by_user",
|
||||||
self._get_devices_with_keys_by_user_txn, user_id,
|
self._get_devices_with_keys_by_user_txn, user_id,
|
||||||
|
@ -380,6 +408,8 @@ class DeviceStore(SQLBaseStore):
|
||||||
return now_stream_id, []
|
return now_stream_id, []
|
||||||
|
|
||||||
def mark_as_sent_devices_by_remote(self, destination, stream_id):
|
def mark_as_sent_devices_by_remote(self, destination, stream_id):
|
||||||
|
"""Mark that updates have successfully been sent to the destination.
|
||||||
|
"""
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
"mark_as_sent_devices_by_remote", self._mark_as_sent_devices_by_remote_txn,
|
"mark_as_sent_devices_by_remote", self._mark_as_sent_devices_by_remote_txn,
|
||||||
destination, stream_id,
|
destination, stream_id,
|
||||||
|
@ -403,6 +433,8 @@ class DeviceStore(SQLBaseStore):
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def get_user_whose_devices_changed(self, from_key):
|
def get_user_whose_devices_changed(self, from_key):
|
||||||
|
"""Get set of users whose devices have changed since `from_key`.
|
||||||
|
"""
|
||||||
from_key = int(from_key)
|
from_key = int(from_key)
|
||||||
changed = self._device_list_stream_cache.get_all_entities_changed(from_key)
|
changed = self._device_list_stream_cache.get_all_entities_changed(from_key)
|
||||||
if changed is not None:
|
if changed is not None:
|
||||||
|
@ -416,8 +448,9 @@ class DeviceStore(SQLBaseStore):
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def add_device_change_to_streams(self, user_id, device_ids, hosts):
|
def add_device_change_to_streams(self, user_id, device_ids, hosts):
|
||||||
# device_lists_stream
|
"""Persist that a user's devices have been updated, and which hosts
|
||||||
# device_lists_outbound_pokes
|
(if any) should be poked.
|
||||||
|
"""
|
||||||
with self._device_list_id_gen.get_next() as stream_id:
|
with self._device_list_id_gen.get_next() as stream_id:
|
||||||
yield self.runInteraction(
|
yield self.runInteraction(
|
||||||
"add_device_change_to_streams", self._add_device_change_txn,
|
"add_device_change_to_streams", self._add_device_change_txn,
|
||||||
|
|
|
@ -13,7 +13,7 @@
|
||||||
* limitations under the License.
|
* limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
|
-- Cache of remote devices.
|
||||||
CREATE TABLE device_lists_remote_cache (
|
CREATE TABLE device_lists_remote_cache (
|
||||||
user_id TEXT NOT NULL,
|
user_id TEXT NOT NULL,
|
||||||
device_id TEXT NOT NULL,
|
device_id TEXT NOT NULL,
|
||||||
|
@ -23,6 +23,8 @@ CREATE TABLE device_lists_remote_cache (
|
||||||
CREATE INDEX device_lists_remote_cache_id ON device_lists_remote_cache(user_id, device_id);
|
CREATE INDEX device_lists_remote_cache_id ON device_lists_remote_cache(user_id, device_id);
|
||||||
|
|
||||||
|
|
||||||
|
-- The last update we got for a user. Empty if we're not receiving updates for
|
||||||
|
-- that user.
|
||||||
CREATE TABLE device_lists_remote_extremeties (
|
CREATE TABLE device_lists_remote_extremeties (
|
||||||
user_id TEXT NOT NULL,
|
user_id TEXT NOT NULL,
|
||||||
stream_id TEXT NOT NULL
|
stream_id TEXT NOT NULL
|
||||||
|
@ -31,6 +33,7 @@ CREATE TABLE device_lists_remote_extremeties (
|
||||||
CREATE INDEX device_lists_remote_extremeties_id ON device_lists_remote_extremeties(user_id, stream_id);
|
CREATE INDEX device_lists_remote_extremeties_id ON device_lists_remote_extremeties(user_id, stream_id);
|
||||||
|
|
||||||
|
|
||||||
|
-- Stream of device lists updates. Includes both local and remotes
|
||||||
CREATE TABLE device_lists_stream (
|
CREATE TABLE device_lists_stream (
|
||||||
stream_id BIGINT NOT NULL,
|
stream_id BIGINT NOT NULL,
|
||||||
user_id TEXT NOT NULL,
|
user_id TEXT NOT NULL,
|
||||||
|
@ -40,6 +43,9 @@ CREATE TABLE device_lists_stream (
|
||||||
CREATE INDEX device_lists_stream_id ON device_lists_stream(stream_id, user_id);
|
CREATE INDEX device_lists_stream_id ON device_lists_stream(stream_id, user_id);
|
||||||
|
|
||||||
|
|
||||||
|
-- The stream of updates to send to other servers. We keep at least one row
|
||||||
|
-- per user that was sent so that the prev_id for any new updates can be
|
||||||
|
-- calculated
|
||||||
CREATE TABLE device_lists_outbound_pokes (
|
CREATE TABLE device_lists_outbound_pokes (
|
||||||
destination TEXT NOT NULL,
|
destination TEXT NOT NULL,
|
||||||
stream_id BIGINT NOT NULL,
|
stream_id BIGINT NOT NULL,
|
||||||
|
|
Loading…
Reference in a new issue