diff --git a/synapse/handlers/__init__.py b/synapse/handlers/__init__.py
index 87b4d381c7..6a2339f2eb 100644
--- a/synapse/handlers/__init__.py
+++ b/synapse/handlers/__init__.py
@@ -17,7 +17,7 @@ from synapse.appservice.scheduler import AppServiceScheduler
from synapse.appservice.api import ApplicationServiceApi
from .register import RegistrationHandler
from .room import (
- RoomCreationHandler, RoomMemberHandler, RoomListHandler
+ RoomCreationHandler, RoomMemberHandler, RoomListHandler, RoomContextHandler,
)
from .message import MessageHandler
from .events import EventStreamHandler, EventHandler
@@ -70,3 +70,4 @@ class Handlers(object):
self.auth_handler = AuthHandler(hs)
self.identity_handler = IdentityHandler(hs)
self.search_handler = SearchHandler(hs)
+ self.room_context_handler = RoomContextHandler(hs)
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 60f9fa58b0..36878a6c20 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -33,6 +33,7 @@ from collections import OrderedDict
from unpaddedbase64 import decode_base64
import logging
+import math
import string
logger = logging.getLogger(__name__)
@@ -747,6 +748,60 @@ class RoomListHandler(BaseHandler):
defer.returnValue({"start": "START", "end": "END", "chunk": chunk})
+class RoomContextHandler(BaseHandler):
+ @defer.inlineCallbacks
+ def get_event_context(self, user, room_id, event_id, limit):
+ """Retrieves events, pagination tokens and state around a given event
+ in a room.
+
+ Args:
+ user (UserID)
+ room_id (str)
+ event_id (str)
+ limit (int): The maximum number of events to return in total
+ (excluding state).
+
+ Returns:
+ dict
+ """
+ before_limit = math.floor(limit/2.)
+ after_limit = limit - before_limit
+
+ now_token = yield self.hs.get_event_sources().get_current_token()
+
+ results = yield self.store.get_events_around(
+ room_id, event_id, before_limit, after_limit
+ )
+
+ results["events_before"] = yield self._filter_events_for_client(
+ user.to_string(), results["events_before"]
+ )
+
+ results["events_after"] = yield self._filter_events_for_client(
+ user.to_string(), results["events_after"]
+ )
+
+ if results["events_after"]:
+ last_event_id = results["events_after"][-1].event_id
+ else:
+ last_event_id = event_id
+
+ state = yield self.store.get_state_for_events(
+ [last_event_id], None
+ )
+ results["state"] = state[last_event_id].values()
+
+ results["start"] = now_token.copy_and_replace(
+ "room_key", results["start"]
+ ).to_string()
+
+ results["end"] = now_token.copy_and_replace(
+ "room_key", results["end"]
+ ).to_string()
+
+ defer.returnValue(results)
+
+
class RoomEventSource(object):
def __init__(self, hs):
self.store = hs.get_datastore()
diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index bbe82b1425..2718e9482e 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -51,6 +51,17 @@ class SearchHandler(BaseHandler):
"content.body", "content.name", "content.topic",
])
filter_dict = content["search_categories"]["room_events"].get("filter", {})
+ event_context = content["search_categories"]["room_events"].get(
+ "event_context", None
+ )
+
+ if event_context is not None:
+ before_limit = int(event_context.get(
+ "before_limit", 5
+ ))
+ after_limit = int(event_context.get(
+ "after_limit", 5
+ ))
except KeyError:
raise SynapseError(400, "Invalid search query")
@@ -76,14 +87,57 @@ class SearchHandler(BaseHandler):
user.to_string(), filtered_events
)
+ allowed_events.sort(key=lambda e: -rank_map[e.event_id])
+ allowed_events = allowed_events[:search_filter.limit()]
+
+ if event_context is not None:
+ now_token = yield self.hs.get_event_sources().get_current_token()
+
+ contexts = {}
+ for event in allowed_events:
+ res = yield self.store.get_events_around(
+ event.room_id, event.event_id, before_limit, after_limit
+ )
+
+ res["events_before"] = yield self._filter_events_for_client(
+ user.to_string(), res["events_before"]
+ )
+
+ res["events_after"] = yield self._filter_events_for_client(
+ user.to_string(), res["events_after"]
+ )
+
+ res["start"] = now_token.copy_and_replace(
+ "room_key", res["start"]
+ ).to_string()
+
+ res["end"] = now_token.copy_and_replace(
+ "room_key", res["end"]
+ ).to_string()
+
+ contexts[event.event_id] = res
+ else:
+ contexts = {}
+
# TODO: Add a limit
time_now = self.clock.time_msec()
+ for context in contexts.values():
+ context["events_before"] = [
+ serialize_event(e, time_now)
+ for e in context["events_before"]
+ ]
+ context["events_after"] = [
+ serialize_event(e, time_now)
+ for e in context["events_after"]
+ ]
+
results = {
e.event_id: {
"rank": rank_map[e.event_id],
- "result": serialize_event(e, time_now)
+ "result": serialize_event(e, time_now),
+ "context": contexts.get(e.event_id, {}),
}
for e in allowed_events
}
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index b8e2c81969..4c5a2353b2 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -113,15 +113,20 @@ class SyncHandler(BaseHandler):
self.clock = hs.get_clock()
@defer.inlineCallbacks
- def wait_for_sync_for_user(self, sync_config, since_token=None, timeout=0):
+ def wait_for_sync_for_user(self, sync_config, since_token=None, timeout=0,
+ full_state=False):
"""Get the sync for a client if we have new data for it now. Otherwise
wait for new data to arrive on the server. If the timeout expires, then
return an empty sync result.
Returns:
A Deferred SyncResult.
"""
- if timeout == 0 or since_token is None:
- result = yield self.current_sync_for_user(sync_config, since_token)
+
+ if timeout == 0 or since_token is None or full_state:
+ # we are going to return immediately, so don't bother calling
+ # notifier.wait_for_events.
+ result = yield self.current_sync_for_user(sync_config, since_token,
+ full_state=full_state)
defer.returnValue(result)
else:
def current_sync_callback(before_token, after_token):
@@ -146,19 +151,24 @@ class SyncHandler(BaseHandler):
)
defer.returnValue(result)
- def current_sync_for_user(self, sync_config, since_token=None):
+ def current_sync_for_user(self, sync_config, since_token=None,
+ full_state=False):
"""Get the sync for client needed to match what the server has now.
Returns:
A Deferred SyncResult.
"""
- if since_token is None:
- return self.initial_sync(sync_config)
+ if since_token is None or full_state:
+ return self.full_state_sync(sync_config, since_token)
else:
return self.incremental_sync_with_gap(sync_config, since_token)
@defer.inlineCallbacks
- def initial_sync(self, sync_config):
- """Get a sync for a client which is starting without any state
+ def full_state_sync(self, sync_config, timeline_since_token):
+ """Get a sync for a client which is starting without any state.
+
+ If a 'message_since_token' is given, only timeline events which have
+ happened since that token will be returned.
+
Returns:
A Deferred SyncResult.
"""
@@ -192,8 +202,12 @@ class SyncHandler(BaseHandler):
archived = []
for event in room_list:
if event.membership == Membership.JOIN:
- room_sync = yield self.initial_sync_for_joined_room(
- event.room_id, sync_config, now_token, typing_by_room
+ room_sync = yield self.full_state_sync_for_joined_room(
+ room_id=event.room_id,
+ sync_config=sync_config,
+ now_token=now_token,
+ timeline_since_token=timeline_since_token,
+ typing_by_room=typing_by_room
)
joined.append(room_sync)
elif event.membership == Membership.INVITE:
@@ -206,11 +220,12 @@ class SyncHandler(BaseHandler):
leave_token = now_token.copy_and_replace(
"room_key", "s%d" % (event.stream_ordering,)
)
- room_sync = yield self.initial_sync_for_archived_room(
+ room_sync = yield self.full_state_sync_for_archived_room(
sync_config=sync_config,
room_id=event.room_id,
leave_event_id=event.event_id,
leave_token=leave_token,
+ timeline_since_token=timeline_since_token,
)
archived.append(room_sync)
@@ -223,15 +238,16 @@ class SyncHandler(BaseHandler):
))
@defer.inlineCallbacks
- def initial_sync_for_joined_room(self, room_id, sync_config, now_token,
- typing_by_room):
+ def full_state_sync_for_joined_room(self, room_id, sync_config,
+ now_token, timeline_since_token,
+ typing_by_room):
"""Sync a room for a client which is starting without any state
Returns:
A Deferred JoinedSyncResult.
"""
batch = yield self.load_filtered_recents(
- room_id, sync_config, now_token,
+ room_id, sync_config, now_token, since_token=timeline_since_token
)
current_state = yield self.state_handler.get_current_state(
@@ -278,15 +294,16 @@ class SyncHandler(BaseHandler):
defer.returnValue((now_token, typing_by_room))
@defer.inlineCallbacks
- def initial_sync_for_archived_room(self, room_id, sync_config,
- leave_event_id, leave_token):
+ def full_state_sync_for_archived_room(self, room_id, sync_config,
+ leave_event_id, leave_token,
+ timeline_since_token):
"""Sync a room for a client which is starting without any state
Returns:
A Deferred JoinedSyncResult.
"""
batch = yield self.load_filtered_recents(
- room_id, sync_config, leave_token,
+ room_id, sync_config, leave_token, since_token=timeline_since_token
)
leave_state = yield self.store.get_state_for_events(
@@ -370,7 +387,7 @@ class SyncHandler(BaseHandler):
else:
prev_batch = now_token
- state = yield self.check_joined_room(
+ state, limited = yield self.check_joined_room(
sync_config, room_id, state
)
@@ -379,7 +396,7 @@ class SyncHandler(BaseHandler):
timeline=TimelineBatch(
events=recents,
prev_batch=prev_batch,
- limited=False,
+ limited=limited,
),
state=state,
ephemeral=typing_by_room.get(room_id, [])
@@ -503,7 +520,7 @@ class SyncHandler(BaseHandler):
current_state=current_state_events,
)
- state_events_delta = yield self.check_joined_room(
+ state_events_delta, _ = yield self.check_joined_room(
sync_config, room_id, state_events_delta
)
@@ -610,6 +627,7 @@ class SyncHandler(BaseHandler):
@defer.inlineCallbacks
def check_joined_room(self, sync_config, room_id, state_delta):
joined = False
+ limited = False
for event in state_delta:
if (
event.type == EventTypes.Member
@@ -621,5 +639,6 @@ class SyncHandler(BaseHandler):
if joined:
res = yield self.state_handler.get_current_state(room_id)
state_delta = res.values()
+ limited = True
- defer.returnValue(state_delta)
+ defer.returnValue((state_delta, limited))
|
