mirror of
https://git.anonymousland.org/anonymousland/synapse.git
synced 2024-09-08 11:01:54 +00:00
Fix the stream change cache to work over replication
This commit is contained in:
parent
cb98ac261b
commit
2a0159b8ae
@ -16,6 +16,7 @@
|
||||
from ._base import BaseSlavedStore
|
||||
from ._slaved_id_tracker import SlavedIdTracker
|
||||
from synapse.storage import DataStore
|
||||
from synapse.util.caches.stream_change_cache import StreamChangeCache
|
||||
|
||||
|
||||
class SlavedDeviceInboxStore(BaseSlavedStore):
|
||||
@ -24,6 +25,10 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
|
||||
self._device_inbox_id_gen = SlavedIdTracker(
|
||||
db_conn, "device_inbox", "stream_id",
|
||||
)
|
||||
self._device_inbox_stream_cache = StreamChangeCache(
|
||||
"DeviceInboxStreamChangeCache",
|
||||
self._device_inbox_id_gen.get_current_token()
|
||||
)
|
||||
|
||||
get_to_device_stream_token = DataStore.get_to_device_stream_token.__func__
|
||||
get_new_messages_for_device = DataStore.get_new_messages_for_device.__func__
|
||||
@ -38,5 +43,11 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
|
||||
stream = result.get("to_device")
|
||||
if stream:
|
||||
self._device_inbox_id_gen.advance(int(stream["position"]))
|
||||
for row in stream["rows"]:
|
||||
stream_id = row[0]
|
||||
user_id = row[1]
|
||||
self._device_inbox_stream_cache.entity_has_changed(
|
||||
user_id, stream_id
|
||||
)
|
||||
|
||||
return super(SlavedDeviceInboxStore, self).process_replication(result)
|
||||
|
@ -202,7 +202,7 @@ class DataStore(RoomMemberStore, RoomStore,
|
||||
max_value=max_device_inbox_id,
|
||||
)
|
||||
self._device_federation_outbox_stream_cache = StreamChangeCache(
|
||||
"DeviceInboxStreamChangeCache", min_device_outbox_id,
|
||||
"DeviceFederationOutboxStreamChangeCache", min_device_outbox_id,
|
||||
prefilled_cache=device_outbox_prefill,
|
||||
)
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user