From 3a2a5b959cb1f56b26af32e1ad4c1db424279eb7 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 26 Aug 2014 18:57:46 +0100 Subject: WIP: Completely change how event streaming and pagination work. This reflects the change in the underlying storage model. --- synapse/streams/config.py | 72 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 synapse/streams/config.py (limited to 'synapse/streams/config.py') diff --git a/synapse/streams/config.py b/synapse/streams/config.py new file mode 100644 index 0000000000..b6ffbab1e7 --- /dev/null +++ b/synapse/streams/config.py @@ -0,0 +1,72 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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 synapse.api.errors import SynapseError +from synapse.types import StreamToken + +import logging + + +logger = logging.getLogger(__name__) + + +class PaginationConfig(object): + + """A configuration object which stores pagination parameters.""" + + def __init__(self, from_tok=None, to_tok=None, direction='f', limit=0): + self.from_token = StreamToken.from_string(from_tok) if from_tok else None + self.to_token = StreamToken.from_string(to_tok) if to_tok else None + self.direction = 'f' if direction == 'f' else 'b' + self.limit = int(limit) + + @classmethod + def from_request(cls, request, raise_invalid_params=True): + params = { + "direction": 'f', + } + + query_param_mappings = [ # 3-tuple of qp_key, attribute, rules + ("from", "from_tok", lambda x: type(x) == str), + ("to", "to_tok", lambda x: type(x) == str), + ("limit", "limit", lambda x: x.isdigit()), + ("dir", "direction", lambda x: x == 'f' or x == 'b'), + ] + + for qp, attr, is_valid in query_param_mappings: + if qp in request.args: + if is_valid(request.args[qp][0]): + params[attr] = request.args[qp][0] + elif raise_invalid_params: + raise SynapseError(400, "%s parameter is invalid." % qp) + + if "from_tok" in params and params["from_tok"] == "END": + # TODO (erikj): This is for compatibility only. + del params["from_tok"] + + try: + return PaginationConfig(**params) + except: + logger.exception("Failed to create pagination config") + raise SynapseError(400, "Invalid request.") + + def __str__(self): + return ( + "" + ) % (self.from_tok, self.to_tok, self.direction, self.limit) + + + -- cgit 1.5.1 From 77a255c7c38a6dab80add45632ffc574099566c8 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 27 Aug 2014 14:19:39 +0100 Subject: PEP8 tweaks. --- synapse/handlers/events.py | 1 - synapse/handlers/room.py | 12 +++++++++--- synapse/streams/config.py | 7 +++---- synapse/streams/events.py | 1 - synapse/types.py | 1 - 5 files changed, 12 insertions(+), 10 deletions(-) (limited to 'synapse/streams/config.py') diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index b336b292d3..e08231406d 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -59,7 +59,6 @@ class EventStreamHandler(BaseHandler): ) self._streams_per_user[auth_user] += 1 - if pagin_config.from_token is None: pagin_config.from_token = None diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 19ade10a91..bf66d74548 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -311,12 +311,18 @@ class MessageHandler(BaseHandler): "end": token[1], } - current_state = yield self.store.get_current_state(event.room_id) + current_state = yield self.store.get_current_state( + event.room_id + ) d["state"] = [c.get_dict() for c in current_state] except: logger.exception("Failed to get snapshot") - ret = {"rooms": rooms_ret, "presence": presence, "end": now_token.to_string()} + ret = { + "rooms": rooms_ret, + "presence": presence, + "end": now_token.to_string() + } defer.returnValue(ret) @@ -499,7 +505,7 @@ class RoomMemberHandler(BaseHandler): for entry in member_list ] chunk_data = { - "start": "START", # FIXME (erikj): START is no longer a valid value + "start": "START", # FIXME (erikj): START is no longer valid "end": "END", "chunk": event_list } diff --git a/synapse/streams/config.py b/synapse/streams/config.py index b6ffbab1e7..69c7145a36 100644 --- a/synapse/streams/config.py +++ b/synapse/streams/config.py @@ -27,7 +27,9 @@ class PaginationConfig(object): """A configuration object which stores pagination parameters.""" def __init__(self, from_tok=None, to_tok=None, direction='f', limit=0): - self.from_token = StreamToken.from_string(from_tok) if from_tok else None + self.from_token = ( + StreamToken.from_string(from_tok) if from_tok else None + ) self.to_token = StreamToken.from_string(to_tok) if to_tok else None self.direction = 'f' if direction == 'f' else 'b' self.limit = int(limit) @@ -67,6 +69,3 @@ class PaginationConfig(object): "" ) % (self.from_tok, self.to_tok, self.direction, self.limit) - - - diff --git a/synapse/streams/events.py b/synapse/streams/events.py index 36174a811b..bf48df5b79 100644 --- a/synapse/streams/events.py +++ b/synapse/streams/events.py @@ -171,4 +171,3 @@ class StreamSource(object): class PaginationSource(object): def get_pagination_rows(self, user, from_token, to_token, limit, key): raise NotImplementedError("get_rows") - diff --git a/synapse/types.py b/synapse/types.py index c8936b5758..63154855dd 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -121,7 +121,6 @@ class StreamToken( str(self.presence_key), ]) - def copy_and_replace(self, key, new_value): d = self._asdict() d[key] = new_value -- cgit 1.5.1 From 05672a6a8ca66bd2165217c06c5478a06b0cd952 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 27 Aug 2014 15:25:21 +0100 Subject: Convert get_paginat_rows to use PaginationConfig. This allows people to supply directions. --- synapse/handlers/room.py | 15 +++++------ synapse/streams/config.py | 67 ++++++++++++++++++++++++++++------------------- synapse/streams/events.py | 20 +++++++++----- 3 files changed, 61 insertions(+), 41 deletions(-) (limited to 'synapse/streams/config.py') diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index bf66d74548..a32c22db33 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -22,6 +22,7 @@ from synapse.api.errors import RoomError, StoreError, SynapseError from synapse.api.events.room import ( RoomTopicEvent, RoomMemberEvent, RoomConfigEvent ) +from synapse.streams.config import PaginationConfig from synapse.util import stringutils from ._base import BaseHandler @@ -115,21 +116,18 @@ class MessageHandler(BaseHandler): data_source = self.hs.get_event_sources().sources[0] - if pagin_config.from_token: - from_token = pagin_config.from_token - else: - from_token = yield self.hs.get_event_sources().get_current_token() + if not pagin_config.from_token: + pagin_config.from_token = yield self.hs.get_event_sources().get_current_token() user = self.hs.parse_userid(user_id) events, next_token = yield data_source.get_pagination_rows( - user, from_token, pagin_config.to_token, pagin_config.limit, - room_id + user, pagin_config, room_id ) chunk = { "chunk": [e.get_dict() for e in events], - "start": from_token.to_string(), + "start": pagin_config.from_token.to_string(), "end": next_token.to_string(), } @@ -277,8 +275,9 @@ class MessageHandler(BaseHandler): # FIXME (erikj): Fix this. presence_stream = self.hs.get_event_sources().sources[1] + pagination_config = PaginationConfig(from_token=now_token) presence, _ = yield presence_stream.get_pagination_rows( - user, now_token, None, None, None + user, pagination_config, None ) limit = pagin_config.limit diff --git a/synapse/streams/config.py b/synapse/streams/config.py index 69c7145a36..2434844d80 100644 --- a/synapse/streams/config.py +++ b/synapse/streams/config.py @@ -26,40 +26,53 @@ class PaginationConfig(object): """A configuration object which stores pagination parameters.""" - def __init__(self, from_tok=None, to_tok=None, direction='f', limit=0): - self.from_token = ( - StreamToken.from_string(from_tok) if from_tok else None - ) - self.to_token = StreamToken.from_string(to_tok) if to_tok else None + def __init__(self, from_token=None, to_token=None, direction='f', + limit=0): + self.from_token = from_token + self.to_token = to_token self.direction = 'f' if direction == 'f' else 'b' self.limit = int(limit) @classmethod def from_request(cls, request, raise_invalid_params=True): - params = { - "direction": 'f', - } - - query_param_mappings = [ # 3-tuple of qp_key, attribute, rules - ("from", "from_tok", lambda x: type(x) == str), - ("to", "to_tok", lambda x: type(x) == str), - ("limit", "limit", lambda x: x.isdigit()), - ("dir", "direction", lambda x: x == 'f' or x == 'b'), - ] - - for qp, attr, is_valid in query_param_mappings: - if qp in request.args: - if is_valid(request.args[qp][0]): - params[attr] = request.args[qp][0] - elif raise_invalid_params: - raise SynapseError(400, "%s parameter is invalid." % qp) - - if "from_tok" in params and params["from_tok"] == "END": - # TODO (erikj): This is for compatibility only. - del params["from_tok"] + def get_param(name, default=None): + lst = request.args.get(name, []) + if len(lst) > 1: + raise SynapseError( + 400, "%s must be specified only once" % (name,) + ) + elif len(lst) == 1: + return lst[0] + else: + return default + + direction = get_param("dir", 'f') + if direction not in ['f', 'b']: + raise SynapseError(400, "'dir' parameter is invalid.") + + from_tok = get_param("from") + to_tok = get_param("to") + + try: + if from_tok == "END": + from_tok = None # For backwards compat. + elif from_tok: + from_tok = StreamToken.from_string(from_tok) + except: + raise SynapseError(400, "'from' paramater is invalid") + + try: + if to_tok: + to_tok = StreamToken.from_string(to_tok) + except: + raise SynapseError(400, "'to' paramater is invalid") + + limit = get_param("limit", "0") + if not limit.isdigit(): + raise SynapseError(400, "'limit' parameter must be an integer.") try: - return PaginationConfig(**params) + return PaginationConfig(from_tok, to_tok, direction, limit) except: logger.exception("Failed to create pagination config") raise SynapseError(400, "Invalid request.") diff --git a/synapse/streams/events.py b/synapse/streams/events.py index bf48df5b79..8a84a9d392 100644 --- a/synapse/streams/events.py +++ b/synapse/streams/events.py @@ -47,14 +47,19 @@ class RoomEventSource(object): return self.store.get_room_events_max_id() @defer.inlineCallbacks - def get_pagination_rows(self, user, from_token, to_token, limit, key): + def get_pagination_rows(self, user, pagination_config, key): + from_token = pagination_config.from_token + to_token = pagination_config.to_token + limit = pagination_config.limit + direction = pagination_config.direction + to_key = to_token.events_key if to_token else None events, next_key = yield self.store.paginate_room_events( room_id=key, from_key=from_token.events_key, to_key=to_key, - direction='b', + direction=direction, limit=limit, with_feedback=True ) @@ -101,7 +106,12 @@ class PresenceSource(object): presence = self.hs.get_handlers().presence_handler return presence._user_cachemap_latest_serial - def get_pagination_rows(self, user, from_token, to_token, limit, key): + def get_pagination_rows(self, user, pagination_config, key): + # TODO (erikj): Does this make sense? Ordering? + + from_token = pagination_config.from_token + to_token = pagination_config.to_token + from_key = int(from_token.presence_key) if to_token: @@ -167,7 +177,5 @@ class StreamSource(object): def get_current_token_part(self): raise NotImplementedError("get_current_token_part") - -class PaginationSource(object): - def get_pagination_rows(self, user, from_token, to_token, limit, key): + def get_pagination_rows(self, user, pagination_config, key): raise NotImplementedError("get_rows") -- cgit 1.5.1