diff --git a/synapse/api/streams/__init__.py b/synapse/api/streams/__init__.py
index d831eafbab..0ba4783ea2 100644
--- a/synapse/api/streams/__init__.py
+++ b/synapse/api/streams/__init__.py
@@ -14,6 +14,7 @@
# limitations under the License.
from synapse.api.errors import SynapseError
+from synapse.types import StreamToken
class PaginationConfig(object):
@@ -21,10 +22,10 @@ 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_tok = from_tok
- self.to_tok = to_tok
- self.direction = direction
- self.limit = limit
+ self.from_tok = StreamToken(from_tok) if from_tok else None
+ self.to_tok = StreamToken(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):
@@ -47,7 +48,10 @@ class PaginationConfig(object):
elif raise_invalid_params:
raise SynapseError(400, "%s parameter is invalid." % qp)
- return PaginationConfig(**params)
+ try:
+ return PaginationConfig(**params)
+ except:
+ raise SynapseError(400, "Invalid request.")
def __str__(self):
return (
diff --git a/synapse/types.py b/synapse/types.py
index fd6a3d1d72..baec8a6002 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -92,3 +92,77 @@ class RoomAlias(DomainSpecificString):
class RoomID(DomainSpecificString):
"""Structure representing a room id. """
SIGIL = "!"
+
+
+class StreamToken(
+ namedtuple(
+ "Token",
+ ("events_type", "topological_key", "stream_key", "presence_key")
+ )
+):
+ _SEPARATOR = "_"
+
+ _TOPOLOGICAL_PREFIX = "t"
+ _STREAM_PREFIX = "s"
+
+ _TOPOLOGICAL_SEPERATOR = "-"
+
+ TOPOLOGICAL_TYPE = "topo"
+ STREAM_TYPE = "stream"
+
+ @classmethod
+ def from_string(cls, string):
+ try:
+ events_part, presence_part = string.split(cls._SEPARATOR)
+
+ presence_key = int(presence_part)
+
+ topo_length = len(cls._TOPOLOGICAL_PREFIX)
+ stream_length = len(cls._STREAM_PREFIX)
+ if events_part[:topo_length] == cls._TOPOLOGICAL_PREFIX:
+ # topological event token
+ topo_tok = events_part[topo_length:]
+ topo_key, stream_key = topo_tok.split(
+ cls._TOPOLOGICAL_SEPERATOR, 1
+ )
+
+ topo_key = int(topo_key)
+ stream_key = int(stream_key)
+
+ events_type = cls.TOPOLOGICAL_TYPE
+ elif events_part[:stream_length] == cls._STREAM_PREFIX:
+ topo_key = None
+ stream_key = int(events_part[stream_length:])
+
+ events_type = cls.STREAM_TYPE
+ else:
+ raise
+
+ return cls(
+ events_type=events_type,
+ topological_key=topo_key,
+ stream_key=stream_key,
+ presence_key=presence_key,
+ )
+ except:
+ raise SynapseError(400, "Invalid Token")
+
+ def to_string(self):
+ if self.events_type == self.TOPOLOGICAL_TYPE:
+ return "".join([
+ self._TOPOLOGICAL_PREFIX,
+ str(self.topological_key),
+ self._TOPOLOGICAL_SEPERATOR,
+ str(self.stream_key),
+ self._SEPARATOR,
+ str(self.presence_key),
+ ])
+ elif self.events_type == self.STREAM_TYPE:
+ return "".join([
+ self._STREAM_PREFIX,
+ str(self.stream_key),
+ self._SEPARATOR,
+ str(self.presence_key),
+ ])
+
+ raise RuntimeError("Unrecognized event type: %s", self.events_type)
|
