diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/app/synchrotron.py | 2 | ||||
| -rw-r--r-- | synapse/handlers/sync.py | 43 | ||||
| -rw-r--r-- | synapse/replication/slave/storage/deviceinbox.py | 30 | ||||
| -rw-r--r-- | synapse/rest/__init__.py | 2 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/sendtodevice.py | 84 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/sync.py | 9 | ||||
| -rw-r--r-- | synapse/storage/__init__.py | 5 | ||||
| -rw-r--r-- | synapse/storage/deviceinbox.py | 140 | ||||
| -rw-r--r-- | synapse/storage/schema/delta/34/device_inbox.sql | 24 | ||||
| -rw-r--r-- | synapse/streams/events.py | 2 | ||||
| -rw-r--r-- | synapse/types.py | 2 |
11 files changed, 339 insertions, 4 deletions
diff --git a/synapse/app/synchrotron.py b/synapse/app/synchrotron.py index e3173533e2..b8cedc7af5 100644 --- a/synapse/app/synchrotron.py +++ b/synapse/app/synchrotron.py @@ -36,6 +36,7 @@ from synapse.replication.slave.storage.registration import SlavedRegistrationSto from synapse.replication.slave.storage.filtering import SlavedFilteringStore from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore from synapse.replication.slave.storage.presence import SlavedPresenceStore +from synapse.replication.slave.storage.deviceinbox import SlavedDeviceInboxStore from synapse.server import HomeServer from synapse.storage.client_ips import ClientIpStore from synapse.storage.engines import create_engine @@ -72,6 +73,7 @@ class SynchrotronSlavedStore( SlavedRegistrationStore, SlavedFilteringStore, SlavedPresenceStore, + SlavedDeviceInboxStore, BaseSlavedStore, ClientIpStore, # After BaseSlavedStore because the constructor is different ): diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 3017bc737b..91934b0c81 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -35,6 +35,7 @@ SyncConfig = collections.namedtuple("SyncConfig", [ "filter_collection", "is_guest", "request_key", + "device_id", ]) @@ -113,6 +114,7 @@ class SyncResult(collections.namedtuple("SyncResult", [ "joined", # JoinedSyncResult for each joined room. "invited", # InvitedSyncResult for each invited room. "archived", # ArchivedSyncResult for each archived room. + "to_device", # List of direct messages for the device. ])): __slots__ = [] @@ -126,7 +128,8 @@ class SyncResult(collections.namedtuple("SyncResult", [ self.joined or self.invited or self.archived or - self.account_data + self.account_data or + self.to_device ) @@ -532,16 +535,53 @@ class SyncHandler(object): sync_result_builder, newly_joined_rooms, newly_joined_users ) + yield self._generate_sync_entry_for_to_device(sync_result_builder) + defer.returnValue(SyncResult( presence=sync_result_builder.presence, account_data=sync_result_builder.account_data, joined=sync_result_builder.joined, invited=sync_result_builder.invited, archived=sync_result_builder.archived, + to_device=sync_result_builder.to_device, next_batch=sync_result_builder.now_token, )) @defer.inlineCallbacks + def _generate_sync_entry_for_to_device(self, sync_result_builder): + """Generates the portion of the sync response. Populates + `sync_result_builder` with the result. + + Args: + sync_result_builder(SyncResultBuilder) + + Returns: + Deferred(dict): A dictionary containing the per room account data. + """ + user_id = sync_result_builder.sync_config.user.to_string() + device_id = sync_result_builder.sync_config.device_id + now_token = sync_result_builder.now_token + since_stream_id = 0 + if sync_result_builder.since_token is not None: + since_stream_id = int(sync_result_builder.since_token.to_device_key) + + if since_stream_id: + logger.debug("Deleting messages up to %d", since_stream_id) + yield self.store.delete_messages_for_device( + user_id, device_id, since_stream_id + ) + + logger.debug("Getting messages up to %d", now_token.to_device_key) + messages, stream_id = yield self.store.get_new_messages_for_device( + user_id, device_id, now_token.to_device_key + ) + logger.debug("Got messages up to %d: %r", stream_id, messages) + sync_result_builder.now_token = now_token.copy_and_replace( + "to_device_key", stream_id + ) + sync_result_builder.to_device = messages + + @defer.inlineCallbacks def _generate_sync_entry_for_account_data(self, sync_result_builder): """Generates the account data portion of the sync response. Populates `sync_result_builder` with the result. @@ -1111,6 +1151,7 @@ class SyncResultBuilder(object): self.joined = [] self.invited = [] self.archived = [] + self.device = [] class RoomSyncResultBuilder(object): diff --git a/synapse/replication/slave/storage/deviceinbox.py b/synapse/replication/slave/storage/deviceinbox.py new file mode 100644 index 0000000000..7583d23708 --- /dev/null +++ b/synapse/replication/slave/storage/deviceinbox.py @@ -0,0 +1,30 @@ +# -*- coding: utf-8 -*- +# Copyright 2016 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from ._base import BaseSlavedStore +from ._slaved_id_tracker import SlavedIdTracker +from synapse.storage import DataStore + + +class SlavedDeviceInboxStore(BaseSlavedStore): + def __init__(self, db_conn, hs): + super(SlavedDeviceInboxStore, self).__init__(db_conn, hs) + self._device_inbox_id_gen = SlavedIdTracker( + db_conn, "device_inbox", "stream_id", + ) + + 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__ diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py index 326780405e..f9f5a3e077 100644 --- a/synapse/rest/__init__.py +++ b/synapse/rest/__init__.py @@ -49,6 +49,7 @@ from synapse.rest.client.v2_alpha import ( notifications, devices, thirdparty, + sendtodevice, ) from synapse.http.server import JsonResource @@ -96,3 +97,4 @@ class ClientRestResource(JsonResource): notifications.register_servlets(hs, client_resource) devices.register_servlets(hs, client_resource) thirdparty.register_servlets(hs, client_resource) + sendtodevice.register_servlets(hs, client_resource) diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py new file mode 100644 index 0000000000..00533741af --- /dev/null +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -0,0 +1,84 @@ +# -*- coding: utf-8 -*- +# Copyright 2016 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging + +from twisted.internet import defer +from synapse.http.servlet import parse_json_object_from_request + +from synapse.http import servlet +from synapse.rest.client.v1.transactions import HttpTransactionStore +from ._base import client_v2_patterns + +logger = logging.getLogger(__name__) + + +class SendToDeviceRestServlet(servlet.RestServlet): + PATTERNS = client_v2_patterns( + "/sendToDevice/(?P<message_type>[^/]*)/(?P<txn_id>[^/]*)$", + releases=[], v2_alpha=False + ) + + def __init__(self, hs): + """ + Args: + hs (synapse.server.HomeServer): server + """ + super(SendToDeviceRestServlet, self).__init__() + self.hs = hs + self.auth = hs.get_auth() + self.store = hs.get_datastore() + self.is_mine_id = hs.is_mine_id + self.txns = HttpTransactionStore() + + @defer.inlineCallbacks + def on_PUT(self, request, message_type, txn_id): + try: + defer.returnValue( + self.txns.get_client_transaction(request, txn_id) + ) + except KeyError: + pass + + requester = yield self.auth.get_user_by_req(request) + + content = parse_json_object_from_request(request) + + # TODO: Prod the notifier to wake up sync streams. + # TODO: Implement replication for the messages. + # TODO: Send the messages to remote servers if needed. + + local_messages = {} + for user_id, by_device in content["messages"].items(): + if self.is_mine_id(user_id): + messages_by_device = { + device_id: { + "content": message_content, + "type": message_type, + "sender": requester.user.to_string(), + } + for device_id, message_content in by_device.items() + } + local_messages[user_id] = messages_by_device + + yield self.store.add_messages_to_device_inbox(local_messages) + + response = (200, {}) + self.txns.store_client_transaction(request, txn_id, response) + defer.returnValue(response) + + +def register_servlets(hs, http_server): + SendToDeviceRestServlet(hs).register(http_server) diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index b11acdbea7..6fc63715aa 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -97,6 +97,7 @@ class SyncRestServlet(RestServlet): request, allow_guest=True ) user = requester.user + device_id = requester.device_id timeout = parse_integer(request, "timeout", default=0) since = parse_string(request, "since") @@ -109,12 +110,12 @@ class SyncRestServlet(RestServlet): logger.info( "/sync: user=%r, timeout=%r, since=%r," - " set_presence=%r, filter_id=%r" % ( - user, timeout, since, set_presence, filter_id + " set_presence=%r, filter_id=%r, device_id=%r" % ( + user, timeout, since, set_presence, filter_id, device_id ) ) - request_key = (user, timeout, since, filter_id, full_state) + request_key = (user, timeout, since, filter_id, full_state, device_id) if filter_id: if filter_id.startswith('{'): @@ -136,6 +137,7 @@ class SyncRestServlet(RestServlet): filter_collection=filter, is_guest=requester.is_guest, request_key=request_key, + device_id=device_id, ) if since is not None: @@ -173,6 +175,7 @@ class SyncRestServlet(RestServlet): response_content = { "account_data": {"events": sync_result.account_data}, + "to_device": {"events": sync_result.to_device}, "presence": self.encode_presence( sync_result.presence, time_now ), diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 7efc5bfeef..4f4c723c5b 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -36,6 +36,7 @@ from .push_rule import PushRuleStore from .media_repository import MediaRepositoryStore from .rejections import RejectionsStore from .event_push_actions import EventPushActionsStore +from .deviceinbox import DeviceInboxStore from .state import StateStore from .signatures import SignatureStore @@ -84,6 +85,7 @@ class DataStore(RoomMemberStore, RoomStore, OpenIdStore, ClientIpStore, DeviceStore, + DeviceInboxStore, ): def __init__(self, db_conn, hs): @@ -108,6 +110,9 @@ class DataStore(RoomMemberStore, RoomStore, self._presence_id_gen = StreamIdGenerator( db_conn, "presence_stream", "stream_id" ) + self._device_inbox_id_gen = StreamIdGenerator( + db_conn, "device_inbox", "stream_id" + ) self._transaction_id_gen = IdGenerator(db_conn, "sent_transactions", "id") self._state_groups_id_gen = StreamIdGenerator(db_conn, "state_groups", "id") diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py new file mode 100644 index 0000000000..e682fe1bd1 --- /dev/null +++ b/synapse/storage/deviceinbox.py @@ -0,0 +1,140 @@ +# -*- coding: utf-8 -*- +# Copyright 2016 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import ujson + +from twisted.internet import defer + +from ._base import SQLBaseStore + + +logger = logging.getLogger(__name__) + + +class DeviceInboxStore(SQLBaseStore): + + @defer.inlineCallbacks + def add_messages_to_device_inbox(self, messages_by_user_then_device): + """ + Args: + messages_by_user_and_device(dict): + Dictionary of user_id to device_id to message. + Returns: + A deferred that resolves when the messages have been inserted. + """ + + def select_devices_txn(txn, user_id, devices): + if not devices: + return [] + sql = ( + "SELECT user_id, device_id FROM devices" + " WHERE user_id = ? AND device_id IN (" + + ",".join("?" * len(devices)) + + ")" + ) + # TODO: Maybe this needs to be done in batches if there are + # too many local devices for a given user. + args = [user_id] + devices + txn.execute(sql, args) + return [tuple(row) for row in txn.fetchall()] + + def add_messages_to_device_inbox_txn(txn, stream_id): + local_users_and_devices = set() + for user_id, messages_by_device in messages_by_user_then_device.items(): + local_users_and_devices.update( + select_devices_txn(txn, user_id, messages_by_device.keys()) + ) + + sql = ( + "INSERT INTO device_inbox" + " (user_id, device_id, stream_id, message_json)" + " VALUES (?,?,?,?)" + ) + rows = [] + for user_id, messages_by_device in messages_by_user_then_device.items(): + for device_id, message in messages_by_device.items(): + message_json = ujson.dumps(message) + # Only insert into the local inbox if the device exists on + # this server + if (user_id, device_id) in local_users_and_devices: + rows.append((user_id, device_id, stream_id, message_json)) + + txn.executemany(sql, rows) + + with self._device_inbox_id_gen.get_next() as stream_id: + yield self.runInteraction( + "add_messages_to_device_inbox", + add_messages_to_device_inbox_txn, + stream_id + ) + + def get_new_messages_for_device( + self, user_id, device_id, current_stream_id, limit=100 + ): + """ + Args: + user_id(str): The recipient user_id. + device_id(str): The recipient device_id. + current_stream_id(int): The current position of the to device + message stream. + Returns: + Deferred ([dict], int): List of messages for the device and where + in the stream the messages got to. + """ + def get_new_messages_for_device_txn(txn): + sql = ( + "SELECT stream_id, message_json FROM device_inbox" + " WHERE user_id = ? AND device_id = ?" + " AND stream_id <= ?" + " ORDER BY stream_id ASC" + " LIMIT ?" + ) + txn.execute(sql, (user_id, device_id, current_stream_id, limit)) + messages = [] + for row in txn.fetchall(): + stream_pos = row[0] + messages.append(ujson.loads(row[1])) + if len(messages) < limit: + stream_pos = current_stream_id + return (messages, stream_pos) + + return self.runInteraction( + "get_new_messages_for_device", get_new_messages_for_device_txn, + ) + + def delete_messages_for_device(self, user_id, device_id, up_to_stream_id): + """ + Args: + user_id(str): The recipient user_id. + device_id(str): The recipient device_id. + up_to_stream_id(int): Where to delete messages up to. + Returns: + A deferred that resolves when the messages have been deleted. + """ + def delete_messages_for_device_txn(txn): + sql = ( + "DELETE FROM device_inbox" + " WHERE user_id = ? AND device_id = ?" + " AND stream_id <= ?" + ) + txn.execute(sql, (user_id, device_id, up_to_stream_id)) + + return self.runInteraction( + "delete_messages_for_device", delete_messages_for_device_txn + ) + + def get_to_device_stream_token(self): + return self._device_inbox_id_gen.get_current_token() diff --git a/synapse/storage/schema/delta/34/device_inbox.sql b/synapse/storage/schema/delta/34/device_inbox.sql new file mode 100644 index 0000000000..e68844c74a --- /dev/null +++ b/synapse/storage/schema/delta/34/device_inbox.sql @@ -0,0 +1,24 @@ +/* Copyright 2016 OpenMarket Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +CREATE TABLE device_inbox ( + user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + stream_id BIGINT NOT NULL, + message_json TEXT NOT NULL -- {"type":, "sender":, "content",} +); + +CREATE INDEX device_inbox_user_stream_id ON device_inbox(user_id, device_id, stream_id); +CREATE INDEX device_inbox_stream_id ON device_inbox(stream_id); diff --git a/synapse/streams/events.py b/synapse/streams/events.py index d4c0bb6732..6bf21d6f5e 100644 --- a/synapse/streams/events.py +++ b/synapse/streams/events.py @@ -43,6 +43,7 @@ class EventSources(object): @defer.inlineCallbacks def get_current_token(self, direction='f'): push_rules_key, _ = self.store.get_push_rules_stream_token() + to_device_key = self.store.get_to_device_stream_token() token = StreamToken( room_key=( @@ -61,5 +62,6 @@ class EventSources(object): yield self.sources["account_data"].get_current_key() ), push_rules_key=push_rules_key, + to_device_key=to_device_key, ) defer.returnValue(token) diff --git a/synapse/types.py b/synapse/types.py index 5349b0c450..9d64e8c4de 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -154,6 +154,7 @@ class StreamToken( "receipt_key", "account_data_key", "push_rules_key", + "to_device_key", )) ): _SEPARATOR = "_" @@ -190,6 +191,7 @@ class StreamToken( or (int(other.receipt_key) < int(self.receipt_key)) or (int(other.account_data_key) < int(self.account_data_key)) or (int(other.push_rules_key) < int(self.push_rules_key)) + or (int(other.to_device_key) < int(self.to_device_key)) ) def copy_and_advance(self, key, new_value): |