diff options
author | Mark Haines <mark.haines@matrix.org> | 2016-08-31 10:38:58 +0100 |
---|---|---|
committer | Mark Haines <mark.haines@matrix.org> | 2016-08-31 10:38:58 +0100 |
commit | 1aa3e1d2874370fbb6fa9f4fd2b8a110d81981fc (patch) | |
tree | bf66d9adf23d6611c3128a9fa936c98e8ddf0da2 /synapse/replication/resource.py | |
parent | Merge pull request #1049 from matrix-org/erikj/presence_users_in_room (diff) | |
download | synapse-1aa3e1d2874370fbb6fa9f4fd2b8a110d81981fc.tar.xz |
Add a replication stream for direct to device messages
Diffstat (limited to 'synapse/replication/resource.py')
-rw-r--r-- | synapse/replication/resource.py | 19 |
1 files changed, 18 insertions, 1 deletions
diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py index 84993b33b3..533676814a 100644 --- a/synapse/replication/resource.py +++ b/synapse/replication/resource.py @@ -42,6 +42,7 @@ STREAM_NAMES = ( ("pushers",), ("state",), ("caches",), + ("to_device",), ) @@ -144,6 +145,7 @@ class ReplicationResource(Resource): pushers_token, state_token, caches_token, + int(stream_token.to_device_key), )) @request_handler() @@ -193,6 +195,7 @@ class ReplicationResource(Resource): yield self.pushers(writer, current_token, limit, request_streams) yield self.state(writer, current_token, limit, request_streams) yield self.caches(writer, current_token, limit, request_streams) + yield self.to_device(writer, current_token, limit, request_streams) self.streams(writer, current_token, request_streams) logger.info("Replicated %d rows", writer.total) @@ -398,6 +401,20 @@ class ReplicationResource(Resource): "position", "cache_func", "keys", "invalidation_ts" )) + @defer.inlineCallbacks + def to_device(self, writer, current_token, limit, request_streams): + current_position = current_token.to_device + + to_device = request_streams.get("to_device") + + if to_device is not None: + to_device_rows = yield self.store.get_all_new_device_messages( + to_device, current_position, limit + ) + writer.write_header_and_rows("to_device", to_device_rows, ( + "position", "user_id", "device_id", "message_json" + )) + class _Writer(object): """Writes the streams as a JSON object as the response to the request""" @@ -426,7 +443,7 @@ class _Writer(object): class _ReplicationToken(collections.namedtuple("_ReplicationToken", ( "events", "presence", "typing", "receipts", "account_data", "backfill", - "push_rules", "pushers", "state", "caches", + "push_rules", "pushers", "state", "caches", "to_device", ))): __slots__ = [] |