Convert get_paginat_rows to use PaginationConfig. This allows people to supply directions.

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")