Convert get_paginat_rows to use PaginationConfig. This allows people to supply directions.
This commit is contained in:
parent
77a255c7c3
commit
05672a6a8c
|
@ -22,6 +22,7 @@ from synapse.api.errors import RoomError, StoreError, SynapseError
|
||||||
from synapse.api.events.room import (
|
from synapse.api.events.room import (
|
||||||
RoomTopicEvent, RoomMemberEvent, RoomConfigEvent
|
RoomTopicEvent, RoomMemberEvent, RoomConfigEvent
|
||||||
)
|
)
|
||||||
|
from synapse.streams.config import PaginationConfig
|
||||||
from synapse.util import stringutils
|
from synapse.util import stringutils
|
||||||
from ._base import BaseHandler
|
from ._base import BaseHandler
|
||||||
|
|
||||||
|
@ -115,21 +116,18 @@ class MessageHandler(BaseHandler):
|
||||||
|
|
||||||
data_source = self.hs.get_event_sources().sources[0]
|
data_source = self.hs.get_event_sources().sources[0]
|
||||||
|
|
||||||
if pagin_config.from_token:
|
if not pagin_config.from_token:
|
||||||
from_token = pagin_config.from_token
|
pagin_config.from_token = yield self.hs.get_event_sources().get_current_token()
|
||||||
else:
|
|
||||||
from_token = yield self.hs.get_event_sources().get_current_token()
|
|
||||||
|
|
||||||
user = self.hs.parse_userid(user_id)
|
user = self.hs.parse_userid(user_id)
|
||||||
|
|
||||||
events, next_token = yield data_source.get_pagination_rows(
|
events, next_token = yield data_source.get_pagination_rows(
|
||||||
user, from_token, pagin_config.to_token, pagin_config.limit,
|
user, pagin_config, room_id
|
||||||
room_id
|
|
||||||
)
|
)
|
||||||
|
|
||||||
chunk = {
|
chunk = {
|
||||||
"chunk": [e.get_dict() for e in events],
|
"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(),
|
"end": next_token.to_string(),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -277,8 +275,9 @@ class MessageHandler(BaseHandler):
|
||||||
|
|
||||||
# FIXME (erikj): Fix this.
|
# FIXME (erikj): Fix this.
|
||||||
presence_stream = self.hs.get_event_sources().sources[1]
|
presence_stream = self.hs.get_event_sources().sources[1]
|
||||||
|
pagination_config = PaginationConfig(from_token=now_token)
|
||||||
presence, _ = yield presence_stream.get_pagination_rows(
|
presence, _ = yield presence_stream.get_pagination_rows(
|
||||||
user, now_token, None, None, None
|
user, pagination_config, None
|
||||||
)
|
)
|
||||||
|
|
||||||
limit = pagin_config.limit
|
limit = pagin_config.limit
|
||||||
|
|
|
@ -26,40 +26,53 @@ class PaginationConfig(object):
|
||||||
|
|
||||||
"""A configuration object which stores pagination parameters."""
|
"""A configuration object which stores pagination parameters."""
|
||||||
|
|
||||||
def __init__(self, from_tok=None, to_tok=None, direction='f', limit=0):
|
def __init__(self, from_token=None, to_token=None, direction='f',
|
||||||
self.from_token = (
|
limit=0):
|
||||||
StreamToken.from_string(from_tok) if from_tok else None
|
self.from_token = from_token
|
||||||
)
|
self.to_token = to_token
|
||||||
self.to_token = StreamToken.from_string(to_tok) if to_tok else None
|
|
||||||
self.direction = 'f' if direction == 'f' else 'b'
|
self.direction = 'f' if direction == 'f' else 'b'
|
||||||
self.limit = int(limit)
|
self.limit = int(limit)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_request(cls, request, raise_invalid_params=True):
|
def from_request(cls, request, raise_invalid_params=True):
|
||||||
params = {
|
def get_param(name, default=None):
|
||||||
"direction": 'f',
|
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
|
||||||
|
|
||||||
query_param_mappings = [ # 3-tuple of qp_key, attribute, rules
|
direction = get_param("dir", 'f')
|
||||||
("from", "from_tok", lambda x: type(x) == str),
|
if direction not in ['f', 'b']:
|
||||||
("to", "to_tok", lambda x: type(x) == str),
|
raise SynapseError(400, "'dir' parameter is invalid.")
|
||||||
("limit", "limit", lambda x: x.isdigit()),
|
|
||||||
("dir", "direction", lambda x: x == 'f' or x == 'b'),
|
|
||||||
]
|
|
||||||
|
|
||||||
for qp, attr, is_valid in query_param_mappings:
|
from_tok = get_param("from")
|
||||||
if qp in request.args:
|
to_tok = get_param("to")
|
||||||
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:
|
try:
|
||||||
return PaginationConfig(**params)
|
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(from_tok, to_tok, direction, limit)
|
||||||
except:
|
except:
|
||||||
logger.exception("Failed to create pagination config")
|
logger.exception("Failed to create pagination config")
|
||||||
raise SynapseError(400, "Invalid request.")
|
raise SynapseError(400, "Invalid request.")
|
||||||
|
|
|
@ -47,14 +47,19 @@ class RoomEventSource(object):
|
||||||
return self.store.get_room_events_max_id()
|
return self.store.get_room_events_max_id()
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@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
|
to_key = to_token.events_key if to_token else None
|
||||||
|
|
||||||
events, next_key = yield self.store.paginate_room_events(
|
events, next_key = yield self.store.paginate_room_events(
|
||||||
room_id=key,
|
room_id=key,
|
||||||
from_key=from_token.events_key,
|
from_key=from_token.events_key,
|
||||||
to_key=to_key,
|
to_key=to_key,
|
||||||
direction='b',
|
direction=direction,
|
||||||
limit=limit,
|
limit=limit,
|
||||||
with_feedback=True
|
with_feedback=True
|
||||||
)
|
)
|
||||||
|
@ -101,7 +106,12 @@ class PresenceSource(object):
|
||||||
presence = self.hs.get_handlers().presence_handler
|
presence = self.hs.get_handlers().presence_handler
|
||||||
return presence._user_cachemap_latest_serial
|
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)
|
from_key = int(from_token.presence_key)
|
||||||
|
|
||||||
if to_token:
|
if to_token:
|
||||||
|
@ -167,7 +177,5 @@ class StreamSource(object):
|
||||||
def get_current_token_part(self):
|
def get_current_token_part(self):
|
||||||
raise NotImplementedError("get_current_token_part")
|
raise NotImplementedError("get_current_token_part")
|
||||||
|
|
||||||
|
def get_pagination_rows(self, user, pagination_config, key):
|
||||||
class PaginationSource(object):
|
|
||||||
def get_pagination_rows(self, user, from_token, to_token, limit, key):
|
|
||||||
raise NotImplementedError("get_rows")
|
raise NotImplementedError("get_rows")
|
||||||
|
|
Loading…
Reference in New Issue