summary refs log tree commit diff
path: root/synapse/replication/slave/storage
diff options
context:
space:
mode:
authorMark Haines <mark.haines@matrix.org>2016-08-31 10:38:58 +0100
committerMark Haines <mark.haines@matrix.org>2016-08-31 10:38:58 +0100
commit1aa3e1d2874370fbb6fa9f4fd2b8a110d81981fc (patch)
treebf66d9adf23d6611c3128a9fa936c98e8ddf0da2 /synapse/replication/slave/storage
parentMerge pull request #1049 from matrix-org/erikj/presence_users_in_room (diff)
downloadsynapse-1aa3e1d2874370fbb6fa9f4fd2b8a110d81981fc.tar.xz
Add a replication stream for direct to device messages
Diffstat (limited to 'synapse/replication/slave/storage')
-rw-r--r--synapse/replication/slave/storage/deviceinbox.py12
1 files changed, 12 insertions, 0 deletions
diff --git a/synapse/replication/slave/storage/deviceinbox.py b/synapse/replication/slave/storage/deviceinbox.py
index 7583d23708..64d8eb2af1 100644
--- a/synapse/replication/slave/storage/deviceinbox.py
+++ b/synapse/replication/slave/storage/deviceinbox.py
@@ -28,3 +28,15 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
     get_to_device_stream_token = DataStore.get_to_device_stream_token.__func__
     get_new_messages_for_device = DataStore.get_new_messages_for_device.__func__
     delete_messages_for_device = DataStore.delete_messages_for_device.__func__
+
+    def stream_positions(self):
+        result = super(SlavedDeviceInboxStore, self).stream_positions()
+        result["to_device"] = self._device_inbox_id_gen.get_current_token()
+        return result
+
+    def process_replication(self, result):
+        stream = result.get("to_device")
+        if stream:
+            self._device_inbox_id_gen.advance(int(stream["position"]))
+
+        return super(SlavedDeviceInboxStore, self).process_replication(result)