|
|
@ -216,11 +216,12 @@ class ReplicationStreamer(object): |
|
|
|
self.federation_sender.federation_ack(token) |
|
|
|
self.federation_sender.federation_ack(token) |
|
|
|
|
|
|
|
|
|
|
|
@measure_func("repl.on_user_sync") |
|
|
|
@measure_func("repl.on_user_sync") |
|
|
|
|
|
|
|
@defer.inlineCallbacks |
|
|
|
def on_user_sync(self, conn_id, user_id, is_syncing, last_sync_ms): |
|
|
|
def on_user_sync(self, conn_id, user_id, is_syncing, last_sync_ms): |
|
|
|
"""A client has started/stopped syncing on a worker. |
|
|
|
"""A client has started/stopped syncing on a worker. |
|
|
|
""" |
|
|
|
""" |
|
|
|
user_sync_counter.inc() |
|
|
|
user_sync_counter.inc() |
|
|
|
self.presence_handler.update_external_syncs_row( |
|
|
|
yield self.presence_handler.update_external_syncs_row( |
|
|
|
conn_id, user_id, is_syncing, last_sync_ms, |
|
|
|
conn_id, user_id, is_syncing, last_sync_ms, |
|
|
|
) |
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
@ -244,11 +245,12 @@ class ReplicationStreamer(object): |
|
|
|
getattr(self.store, cache_func).invalidate(tuple(keys)) |
|
|
|
getattr(self.store, cache_func).invalidate(tuple(keys)) |
|
|
|
|
|
|
|
|
|
|
|
@measure_func("repl.on_user_ip") |
|
|
|
@measure_func("repl.on_user_ip") |
|
|
|
|
|
|
|
@defer.inlineCallbacks |
|
|
|
def on_user_ip(self, user_id, access_token, ip, user_agent, device_id, last_seen): |
|
|
|
def on_user_ip(self, user_id, access_token, ip, user_agent, device_id, last_seen): |
|
|
|
"""The client saw a user request |
|
|
|
"""The client saw a user request |
|
|
|
""" |
|
|
|
""" |
|
|
|
user_ip_cache_counter.inc() |
|
|
|
user_ip_cache_counter.inc() |
|
|
|
self.store.insert_client_ip( |
|
|
|
yield self.store.insert_client_ip( |
|
|
|
user_id, access_token, ip, user_agent, device_id, last_seen, |
|
|
|
user_id, access_token, ip, user_agent, device_id, last_seen, |
|
|
|
) |
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|