OpenCloud
/
MatrixSynapse
mirror of https://github.com/matrix-org/synapse
1
0
Fork 0
Code Issues Releases Wiki Activity

Poke the slaved pushers on new receipts

markjh/split_pusher
Mark Haines 2016-04-14 16:59:56 +01:00
parent f41b1a8723
commit 1c1b2de975
2 changed files with 11 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
synapse/app/pusher.py
View File

@ -136,6 +136,16 @@ class PusherServer(HomeServer):
min_stream_id, max_stream_id min_stream_id, max_stream_id
) )
stream = results.get("receipts")
if stream:
rows = stream["rows"]
affected_room_ids = set(row[1] for row in rows)
min_stream_id = rows[0][0]
max_stream_id = stream["position"]
preserve_fn(pusher_pool.on_new_receipts)(
min_stream_id, max_stream_id, affected_room_ids
)
while True: while True:
try: try:
args = store.stream_positions() args = store.stream_positions()

1
synapse/replication/slave/storage/receipts.py
View File

@ -40,6 +40,7 @@ class SlavedReceiptsStore(BaseSlavedStore):
get_receipts_for_user = ReceiptsStore.__dict__["get_receipts_for_user"] get_receipts_for_user = ReceiptsStore.__dict__["get_receipts_for_user"]
get_max_receipt_stream_id = DataStore.get_max_receipt_stream_id.__func__ get_max_receipt_stream_id = DataStore.get_max_receipt_stream_id.__func__
get_all_updated_receipts = DataStore.get_all_updated_receipts.__func__
def stream_positions(self): def stream_positions(self):
result = super(SlavedReceiptsStore, self).stream_positions() result = super(SlavedReceiptsStore, self).stream_positions()