summary refs log tree commit diff
path: root/synapse/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/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/storage')
-rw-r--r--synapse/storage/deviceinbox.py39
1 files changed, 39 insertions, 0 deletions
diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py
index e682fe1bd1..27ed1004da 100644
--- a/synapse/storage/deviceinbox.py
+++ b/synapse/storage/deviceinbox.py
@@ -136,5 +136,44 @@ class DeviceInboxStore(SQLBaseStore):
             "delete_messages_for_device", delete_messages_for_device_txn
         )
 
+    def get_all_new_device_messages(self, last_pos, current_pos, limit):
+        """
+        Args:
+            last_pos(int):
+            current_pos(int):
+            limit(int):
+        Returns:
+            A deferred list of rows from the device inbox
+        """
+        if last_pos == current_pos:
+            return defer.succeed([])
+
+        def get_all_new_device_messages_txn(txn):
+            sql = (
+                "SELECT stream_id FROM device_inbox"
+                " WHERE ? < stream_id AND stream_id <= ?"
+                " GROUP BY stream_id"
+                " ORDER BY stream_id ASC"
+                " LIMIT ?"
+            )
+            txn.execute(sql, (last_pos, current_pos, limit))
+            stream_ids = txn.fetchall()
+            if not stream_ids:
+                return []
+            max_stream_id_in_limit = stream_ids[-1]
+
+            sql = (
+                "SELECT stream_id, user_id, device_id, message_json"
+                " FROM device_inbox"
+                " WHERE ? < stream_id AND stream_id <= ?"
+                " ORDER BY stream_id ASC"
+            )
+            txn.execute(sql, (last_pos, max_stream_id_in_limit))
+            return txn.fetchall()
+
+        return self.runInteraction(
+            "get_all_new_device_messages", get_all_new_device_messages_txn
+        )
+
     def get_to_device_stream_token(self):
         return self._device_inbox_id_gen.get_current_token()