mirror of
https://mau.dev/maunium/synapse.git
synced 2024-12-21 08:33:59 +01:00
Merge pull request #4797 from matrix-org/rav/inline_rr_send
Clean up read-receipt handling.
This commit is contained in:
commit
cd7110c869
2 changed files with 43 additions and 57 deletions
1
changelog.d/4797.misc
Normal file
1
changelog.d/4797.misc
Normal file
|
@ -0,0 +1 @@
|
||||||
|
Clean up read-receipt handling.
|
|
@ -16,7 +16,6 @@ import logging
|
||||||
|
|
||||||
from twisted.internet import defer
|
from twisted.internet import defer
|
||||||
|
|
||||||
from synapse.metrics.background_process_metrics import run_as_background_process
|
|
||||||
from synapse.types import get_domain_from_id
|
from synapse.types import get_domain_from_id
|
||||||
|
|
||||||
from ._base import BaseHandler
|
from ._base import BaseHandler
|
||||||
|
@ -38,31 +37,6 @@ class ReceiptsHandler(BaseHandler):
|
||||||
self.clock = self.hs.get_clock()
|
self.clock = self.hs.get_clock()
|
||||||
self.state = hs.get_state_handler()
|
self.state = hs.get_state_handler()
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
|
||||||
def received_client_receipt(self, room_id, receipt_type, user_id,
|
|
||||||
event_id):
|
|
||||||
"""Called when a client tells us a local user has read up to the given
|
|
||||||
event_id in the room.
|
|
||||||
"""
|
|
||||||
receipt = {
|
|
||||||
"room_id": room_id,
|
|
||||||
"receipt_type": receipt_type,
|
|
||||||
"user_id": user_id,
|
|
||||||
"event_ids": [event_id],
|
|
||||||
"data": {
|
|
||||||
"ts": int(self.clock.time_msec()),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
is_new = yield self._handle_new_receipts([receipt])
|
|
||||||
|
|
||||||
if is_new:
|
|
||||||
# fire off a process in the background to send the receipt to
|
|
||||||
# remote servers
|
|
||||||
run_as_background_process(
|
|
||||||
'push_receipts_to_remotes', self._push_remotes, receipt
|
|
||||||
)
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def _received_remote_receipt(self, origin, content):
|
def _received_remote_receipt(self, origin, content):
|
||||||
"""Called when we receive an EDU of type m.receipt from a remote HS.
|
"""Called when we receive an EDU of type m.receipt from a remote HS.
|
||||||
|
@ -128,43 +102,54 @@ class ReceiptsHandler(BaseHandler):
|
||||||
defer.returnValue(True)
|
defer.returnValue(True)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def _push_remotes(self, receipt):
|
def received_client_receipt(self, room_id, receipt_type, user_id,
|
||||||
"""Given a receipt, works out which remote servers should be
|
event_id):
|
||||||
poked and pokes them.
|
"""Called when a client tells us a local user has read up to the given
|
||||||
|
event_id in the room.
|
||||||
"""
|
"""
|
||||||
try:
|
receipt = {
|
||||||
# TODO: optimise this to move some of the work to the workers.
|
"room_id": room_id,
|
||||||
room_id = receipt["room_id"]
|
"receipt_type": receipt_type,
|
||||||
receipt_type = receipt["receipt_type"]
|
"user_id": user_id,
|
||||||
user_id = receipt["user_id"]
|
"event_ids": [event_id],
|
||||||
event_ids = receipt["event_ids"]
|
"data": {
|
||||||
data = receipt["data"]
|
"ts": int(self.clock.time_msec()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
users = yield self.state.get_current_user_in_room(room_id)
|
is_new = yield self._handle_new_receipts([receipt])
|
||||||
remotedomains = set(get_domain_from_id(u) for u in users)
|
if not is_new:
|
||||||
remotedomains = remotedomains.copy()
|
return
|
||||||
remotedomains.discard(self.server_name)
|
|
||||||
|
|
||||||
logger.debug("Sending receipt to: %r", remotedomains)
|
# Work out which remote servers should be poked and poke them.
|
||||||
|
|
||||||
for domain in remotedomains:
|
# TODO: optimise this to move some of the work to the workers.
|
||||||
self.federation.build_and_send_edu(
|
data = receipt["data"]
|
||||||
destination=domain,
|
|
||||||
edu_type="m.receipt",
|
# XXX why does this not use state.get_current_hosts_in_room() ?
|
||||||
content={
|
users = yield self.state.get_current_user_in_room(room_id)
|
||||||
room_id: {
|
remotedomains = set(get_domain_from_id(u) for u in users)
|
||||||
receipt_type: {
|
remotedomains = remotedomains.copy()
|
||||||
user_id: {
|
remotedomains.discard(self.server_name)
|
||||||
"event_ids": event_ids,
|
|
||||||
"data": data,
|
logger.debug("Sending receipt to: %r", remotedomains)
|
||||||
}
|
|
||||||
|
for domain in remotedomains:
|
||||||
|
self.federation.build_and_send_edu(
|
||||||
|
destination=domain,
|
||||||
|
edu_type="m.receipt",
|
||||||
|
content={
|
||||||
|
room_id: {
|
||||||
|
receipt_type: {
|
||||||
|
user_id: {
|
||||||
|
"event_ids": [event_id],
|
||||||
|
"data": data,
|
||||||
}
|
}
|
||||||
},
|
}
|
||||||
},
|
},
|
||||||
key=(room_id, receipt_type, user_id),
|
},
|
||||||
)
|
key=(room_id, receipt_type, user_id),
|
||||||
except Exception:
|
)
|
||||||
logger.exception("Error pushing receipts to remote servers")
|
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def get_receipts_for_room(self, room_id, to_key):
|
def get_receipts_for_room(self, room_id, to_key):
|
||||||
|
|
Loading…
Reference in a new issue