Merge pull request #2120 from matrix-org/luke/read-markers

Implement Read Marker API
This commit is contained in:
Luke Barnard 2017-04-13 14:21:31 +01:00 committed by GitHub
commit 78f0ddbfad
6 changed files with 167 additions and 1 deletions

View File

@ -0,0 +1,64 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ._base import BaseHandler
from twisted.internet import defer
from synapse.util.async import Linearizer
import logging
logger = logging.getLogger(__name__)
class ReadMarkerHandler(BaseHandler):
def __init__(self, hs):
super(ReadMarkerHandler, self).__init__(hs)
self.server_name = hs.config.server_name
self.store = hs.get_datastore()
self.read_marker_linearizer = Linearizer(name="read_marker")
self.notifier = hs.get_notifier()
@defer.inlineCallbacks
def received_client_read_marker(self, room_id, user_id, event_id):
"""Updates the read marker for a given user in a given room if the event ID given
is ahead in the stream relative to the current read marker.
This uses a notifier to indicate that account data should be sent down /sync if
the read marker has changed.
"""
with (yield self.read_marker_linearizer.queue((room_id, user_id))):
account_data = yield self.store.get_account_data_for_room(user_id, room_id)
existing_read_marker = account_data.get("m.read_marker", None)
should_update = True
if existing_read_marker:
# Only update if the new marker is ahead in the stream
should_update = yield self.store.is_event_after(
event_id,
existing_read_marker['marker']
)
if should_update:
content = {
"marker": event_id
}
max_id = yield self.store.add_account_data_to_room(
user_id, room_id, "m.read_marker", content
)
self.notifier.on_new_event("account_data_key", max_id, users=[user_id])

View File

@ -40,6 +40,7 @@ from synapse.rest.client.v2_alpha import (
register, register,
auth, auth,
receipts, receipts,
read_marker,
keys, keys,
tokenrefresh, tokenrefresh,
tags, tags,
@ -88,6 +89,7 @@ class ClientRestResource(JsonResource):
register.register_servlets(hs, client_resource) register.register_servlets(hs, client_resource)
auth.register_servlets(hs, client_resource) auth.register_servlets(hs, client_resource)
receipts.register_servlets(hs, client_resource) receipts.register_servlets(hs, client_resource)
read_marker.register_servlets(hs, client_resource)
keys.register_servlets(hs, client_resource) keys.register_servlets(hs, client_resource)
tokenrefresh.register_servlets(hs, client_resource) tokenrefresh.register_servlets(hs, client_resource)
tags.register_servlets(hs, client_resource) tags.register_servlets(hs, client_resource)

View File

@ -16,7 +16,7 @@
from ._base import client_v2_patterns from ._base import client_v2_patterns
from synapse.http.servlet import RestServlet, parse_json_object_from_request from synapse.http.servlet import RestServlet, parse_json_object_from_request
from synapse.api.errors import AuthError from synapse.api.errors import AuthError, SynapseError
from twisted.internet import defer from twisted.internet import defer
@ -82,6 +82,13 @@ class RoomAccountDataServlet(RestServlet):
body = parse_json_object_from_request(request) body = parse_json_object_from_request(request)
if account_data_type == "m.read_marker":
raise SynapseError(
405,
"Cannot set m.read_marker through this API."
" Use /rooms/!roomId:server.name/read_marker"
)
max_id = yield self.store.add_account_data_to_room( max_id = yield self.store.add_account_data_to_room(
user_id, room_id, account_data_type, body user_id, room_id, account_data_type, body
) )

View File

@ -0,0 +1,66 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from twisted.internet import defer
from synapse.http.servlet import RestServlet, parse_json_object_from_request
from ._base import client_v2_patterns
import logging
logger = logging.getLogger(__name__)
class ReadMarkerRestServlet(RestServlet):
PATTERNS = client_v2_patterns("/rooms/(?P<room_id>[^/]*)/read_marker$")
def __init__(self, hs):
super(ReadMarkerRestServlet, self).__init__()
self.auth = hs.get_auth()
self.receipts_handler = hs.get_receipts_handler()
self.read_marker_handler = hs.get_read_marker_handler()
self.presence_handler = hs.get_presence_handler()
@defer.inlineCallbacks
def on_POST(self, request, room_id):
requester = yield self.auth.get_user_by_req(request)
yield self.presence_handler.bump_presence_active_time(requester.user)
body = parse_json_object_from_request(request)
read_event_id = body.get("m.read", None)
if read_event_id:
yield self.receipts_handler.received_client_receipt(
room_id,
"m.read",
user_id=requester.user.to_string(),
event_id=read_event_id
)
read_marker_event_id = body.get("m.read_marker", None)
if read_marker_event_id:
yield self.read_marker_handler.received_client_read_marker(
room_id,
user_id=requester.user.to_string(),
event_id=read_marker_event_id
)
defer.returnValue((200, {}))
def register_servlets(hs, http_server):
ReadMarkerRestServlet(hs).register(http_server)

View File

@ -48,6 +48,7 @@ from synapse.handlers.typing import TypingHandler
from synapse.handlers.events import EventHandler, EventStreamHandler from synapse.handlers.events import EventHandler, EventStreamHandler
from synapse.handlers.initial_sync import InitialSyncHandler from synapse.handlers.initial_sync import InitialSyncHandler
from synapse.handlers.receipts import ReceiptsHandler from synapse.handlers.receipts import ReceiptsHandler
from synapse.handlers.read_marker import ReadMarkerHandler
from synapse.http.client import SimpleHttpClient, InsecureInterceptableContextFactory from synapse.http.client import SimpleHttpClient, InsecureInterceptableContextFactory
from synapse.http.matrixfederationclient import MatrixFederationHttpClient from synapse.http.matrixfederationclient import MatrixFederationHttpClient
from synapse.notifier import Notifier from synapse.notifier import Notifier
@ -133,6 +134,7 @@ class HomeServer(object):
'receipts_handler', 'receipts_handler',
'macaroon_generator', 'macaroon_generator',
'tcp_replication', 'tcp_replication',
'read_marker_handler',
] ]
def __init__(self, hostname, **kwargs): def __init__(self, hostname, **kwargs):
@ -291,6 +293,9 @@ class HomeServer(object):
def build_receipts_handler(self): def build_receipts_handler(self):
return ReceiptsHandler(self) return ReceiptsHandler(self)
def build_read_marker_handler(self):
return ReadMarkerHandler(self)
def build_tcp_replication(self): def build_tcp_replication(self):
raise NotImplementedError() raise NotImplementedError()

View File

@ -2159,6 +2159,28 @@ class EventsStore(SQLBaseStore):
] ]
) )
@defer.inlineCallbacks
def is_event_after(self, event_id1, event_id2):
"""Returns True if event_id1 is after event_id2 in the stream
"""
to_1, so_1 = yield self._get_event_ordering(event_id1)
to_2, so_2 = yield self._get_event_ordering(event_id2)
defer.returnValue((to_1, so_1) > (to_2, so_2))
@defer.inlineCallbacks
def _get_event_ordering(self, event_id):
res = yield self._simple_select_one(
table="events",
retcols=["topological_ordering", "stream_ordering"],
keyvalues={"event_id": event_id},
allow_none=True
)
if not res:
raise SynapseError(404, "Could not find event %s" % (event_id,))
defer.returnValue((int(res["topological_ordering"]), int(res["stream_ordering"])))
AllNewEventsResult = namedtuple("AllNewEventsResult", [ AllNewEventsResult = namedtuple("AllNewEventsResult", [
"new_forward_events", "new_backfill_events", "new_forward_events", "new_backfill_events",