OpenCloud
/
MatrixSynapse
mirror of https://github.com/matrix-org/synapse
1
0
Fork 0
Code Issues Releases Wiki Activity

Add tag handling

erikj/paginate_sync
Erik Johnston 2016-06-22 10:59:24 +01:00
parent cdd379b6df
commit 6a101e512f
4 changed files with 60 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
synapse/handlers/sync.py
View File

@ -21,7 +21,7 @@ from synapse.util.caches.response_cache import ResponseCache
from synapse.push.clientformat import format_push_rules_for_user
from synapse.visibility import filter_events_for_client
from synapse.types import SyncNextBatchToken, SyncPaginationState
from synapse.api.errors import Codes
from synapse.api.errors import Codes, SynapseError
from twisted.internet import defer
@ -41,10 +41,30 @@ SyncConfig = collections.namedtuple("SyncConfig", [
])
SyncPaginationConfig = collections.namedtuple("SyncPaginationConfig", [
class SyncPaginationConfig(collections.namedtuple("SyncPaginationConfig", [
"order",
"limit",
])
"tags",
])):
def __init__(self, order, limit, tags):
if order not in SYNC_PAGINATION_VALID_ORDERS:
raise SynapseError(400, "Invalid 'order'")
if tags not in SYNC_PAGINATION_VALID_TAGS_OPTIONS:
raise SynapseError(400, "Invalid 'tags'")
try:
limit = int(limit)
except:
raise SynapseError(400, "Invalid 'limit'")
super(SyncPaginationConfig, self).__init__(order, limit, tags)
SYNC_PAGINATION_TAGS_INCLUDE_ALL = "include_all"
SYNC_PAGINATION_TAGS_IGNORE = "ignore"
SYNC_PAGINATION_VALID_TAGS_OPTIONS = (
SYNC_PAGINATION_TAGS_INCLUDE_ALL, SYNC_PAGINATION_TAGS_IGNORE,
)
SYNC_PAGINATION_ORDER_TS = "o"
SYNC_PAGINATION_VALID_ORDERS = (SYNC_PAGINATION_ORDER_TS,)
@ -727,15 +747,19 @@ class SyncHandler(object):
if sync_config.pagination_config:
pagination_config = sync_config.pagination_config
old_pagination_value = 0
include_all_tags = pagination_config.tags == SYNC_PAGINATION_TAGS_INCLUDE_ALL
elif sync_result_builder.pagination_state:
pagination_config = SyncPaginationConfig(
order=sync_result_builder.pagination_state.order,
limit=sync_result_builder.pagination_state.limit,
tags=sync_result_builder.pagination_state.tags,
)
old_pagination_value = sync_result_builder.pagination_state.value
include_all_tags = pagination_config.tags == SYNC_PAGINATION_TAGS_INCLUDE_ALL
else:
pagination_config = None
old_pagination_value = 0
include_all_tags = False
include_map = extras.get("peek", {}) if extras else {}
@ -752,19 +776,19 @@ class SyncHandler(object):
if missing_state:
for r in room_entries:
if r.room_id in missing_state:
if r.room_id in all_tags:
if include_all_tags and r.room_id in all_tags:
r.always_include = True
continue
r.full_state = True
r.would_require_resync = True
if r.room_id in include_map:
r.always_include = True
r.events = None
r.since_token = None
r.upto_token = now_token
elif pagination_config:
elif pagination_config and include_all_tags:
all_tags = yield self.store.get_tags_for_user(user_id)
logger.info("all_tags: %r", all_tags)
for r in room_entries:
if r.room_id in all_tags:
r.always_include = True
@ -817,8 +841,15 @@ class SyncHandler(object):
sync_result_builder.pagination_state = SyncPaginationState(
order=pagination_config.order, value=value,
limit=pagination_limit + extra_limit,
tags=pagination_config.tags,
)
to_sync_map = {
key: value for key, value in cutoff_list
}
else:
to_sync_map = {}
sync_result_builder.pagination_info["limited"] = limited
if len(room_map) == len(room_entries):
@ -826,7 +857,7 @@ class SyncHandler(object):
room_entries = [
r for r in room_entries
if r.room_id in room_map or r.always_include
if r.room_id in to_sync_map or r.always_include
]
sync_result_builder.full_state |= sync_result_builder.since_token is None
@ -1094,6 +1125,7 @@ class SyncHandler(object):
room_builder.full_state
or newly_joined
or sync_result_builder.full_state
or room_builder.would_require_resync
)
events = room_builder.events
@ -1139,6 +1171,16 @@ class SyncHandler(object):
if not (always_include or batch or account_data or ephemeral):
return
if room_builder.would_require_resync:
since_token = None
batch = yield self._load_filtered_recents(
room_id, sync_config,
now_token=upto_token,
since_token=since_token,
recents=None,
newly_joined_room=newly_joined,
)
state = yield self.compute_state_delta(
room_id, batch, sync_config, since_token, now_token,
full_state=full_state
@ -1343,7 +1385,7 @@ class RoomSyncResultBuilder(object):
__slots__ = (
"room_id", "rtype", "events", "newly_joined", "full_state", "since_token",
"upto_token", "always_include",
"upto_token", "always_include", "would_require_resync",
)
def __init__(self, room_id, rtype, events, newly_joined, full_state,
@ -1367,3 +1409,4 @@ class RoomSyncResultBuilder(object):
self.since_token = since_token
self.upto_token = upto_token
self.always_include = False
self.would_require_resync = False

2
synapse/python_dependencies.py
View File

@ -36,7 +36,7 @@ REQUIREMENTS = {
"blist": ["blist"],
"pysaml2>=3.0.0,<4.0.0": ["saml2>=3.0.0,<4.0.0"],
"pymacaroons-pynacl": ["pymacaroons"],
"msgpack": ["msgpack"],
"msgpack-python": ["msgpack"],
}
CONDITIONAL_REQUIREMENTS = {
"web_client": {

5
synapse/rest/client/v2_alpha/sync.py
View File

@ -19,7 +19,9 @@ from synapse.http.servlet import (
RestServlet, parse_string, parse_integer, parse_boolean,
parse_json_object_from_request,
)
from synapse.handlers.sync import SyncConfig, SyncPaginationConfig
from synapse.handlers.sync import (
SyncConfig, SyncPaginationConfig, SYNC_PAGINATION_TAGS_INCLUDE_ALL,
)
from synapse.types import SyncNextBatchToken
from synapse.events.utils import (
serialize_event, format_event_for_client_v2_without_room_id,
@ -149,6 +151,7 @@ class SyncRestServlet(RestServlet):
pagination_config=SyncPaginationConfig(
order=pagination_config["order"],
limit=pagination_config["limit"],
tags=pagination_config.get("tags", SYNC_PAGINATION_TAGS_INCLUDE_ALL),
) if pagination_config else None,
)

6
synapse/types.py
View File

@ -18,9 +18,9 @@ from synapse.api.errors import SynapseError
from collections import namedtuple
from unpaddedbase64 import encode_base64, decode_base64
import ujson as json
import msgpack
Requester = namedtuple("Requester", ["user", "access_token_id", "is_guest"])
@ -153,12 +153,13 @@ class SyncPaginationState(
"order",
"value",
"limit",
"tags",
))
):
@classmethod
def from_dict(cls, d):
try:
return cls(d["o"], d["v"], d["l"])
return cls(d["o"], d["v"], d["l"], d["t"])
except:
raise SynapseError(400, "Invalid Token")
@ -167,6 +168,7 @@ class SyncPaginationState(
"o": self.order,
"v": self.value,
"l": self.limit,
"t": self.tags,
}
def replace(self, **kwargs):