diff options
| -rw-r--r-- | synapse/handlers/room.py | 15 | ||||
| -rw-r--r-- | synapse/streams/config.py | 67 | ||||
| -rw-r--r-- | synapse/streams/events.py | 20 |
3 files changed, 61 insertions, 41 deletions
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") |