diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 6bb797caf2..2d7bd5083b 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -16,18 +16,14 @@
from twisted.internet import defer
from ._base import BaseHandler
-from synapse.api.streams.event import (
- EventStream, EventsStreamData
-)
-from synapse.handlers.presence import PresenceStreamData
+import logging
-class EventStreamHandler(BaseHandler):
- stream_data_classes = [
- EventsStreamData,
- PresenceStreamData,
- ]
+logger = logging.getLogger(__name__)
+
+
+class EventStreamHandler(BaseHandler):
def __init__(self, hs):
super(EventStreamHandler, self).__init__(hs)
@@ -43,104 +39,22 @@ class EventStreamHandler(BaseHandler):
self.clock = hs.get_clock()
- def get_event_stream_token(self, stream_type, store_id, start_token):
- """Return the next token after this event.
-
- Args:
- stream_type (str): The StreamData.EVENT_TYPE
- store_id (int): The new storage ID assigned from the data store.
- start_token (str): The token the user started with.
- Returns:
- str: The end token.
- """
- for i, stream_cls in enumerate(EventStreamHandler.stream_data_classes):
- if stream_cls.EVENT_TYPE == stream_type:
- # this is the stream for this event, so replace this part of
- # the token
- store_ids = start_token.split(EventStream.SEPARATOR)
- store_ids[i] = str(store_id)
- return EventStream.SEPARATOR.join(store_ids)
- raise RuntimeError("Didn't find a stream type %s" % stream_type)
+ self.notifier = hs.get_notifier()
@defer.inlineCallbacks
def get_stream(self, auth_user_id, pagin_config, timeout=0):
- """Gets events as an event stream for this user.
-
- This function looks for interesting *events* for this user. This is
- different from the notifier, which looks for interested *users* who may
- want to know about a single event.
-
- Args:
- auth_user_id (str): The user requesting their event stream.
- pagin_config (synapse.api.streams.PaginationConfig): The config to
- use when obtaining the stream.
- timeout (int): The max time to wait for an incoming event in ms.
- Returns:
- A pagination stream API dict
- """
auth_user = self.hs.parse_userid(auth_user_id)
- stream_id = object()
-
- try:
- if auth_user not in self._streams_per_user:
- self._streams_per_user[auth_user] = 0
- if auth_user in self._stop_timer_per_user:
- self.clock.cancel_call_later(
- self._stop_timer_per_user.pop(auth_user))
- else:
- self.distributor.fire(
- "started_user_eventstream", auth_user
- )
- self._streams_per_user[auth_user] += 1
-
- # construct an event stream with the correct data ordering
- stream_data_list = []
- for stream_class in EventStreamHandler.stream_data_classes:
- stream_data_list.append(stream_class(self.hs))
- event_stream = EventStream(auth_user_id, stream_data_list)
-
- # fix unknown tokens to known tokens
- pagin_config = yield event_stream.fix_tokens(pagin_config)
-
- # register interest in receiving new events
- self.notifier.store_events_for(user_id=auth_user_id,
- stream_id=stream_id,
- from_tok=pagin_config.from_tok)
-
- # see if we can grab a chunk now
- data_chunk = yield event_stream.get_chunk(config=pagin_config)
-
- # if there are previous events, return those. If not, wait on the
- # new events for 'timeout' seconds.
- if len(data_chunk["chunk"]) == 0 and timeout != 0:
- results = yield defer.maybeDeferred(
- self.notifier.get_events_for,
- user_id=auth_user_id,
- stream_id=stream_id,
- timeout=timeout
- )
- if results:
- defer.returnValue(results)
-
- defer.returnValue(data_chunk)
- finally:
- # cleanup
- self.notifier.purge_events_for(user_id=auth_user_id,
- stream_id=stream_id)
-
- self._streams_per_user[auth_user] -= 1
- if not self._streams_per_user[auth_user]:
- del self._streams_per_user[auth_user]
-
- # 10 seconds of grace to allow the client to reconnect again
- # before we think they're gone
- def _later():
- self.distributor.fire(
- "stopped_user_eventstream", auth_user
- )
- del self._stop_timer_per_user[auth_user]
-
- self._stop_timer_per_user[auth_user] = (
- self.clock.call_later(5, _later)
- )
+ if pagin_config.from_token is None:
+ pagin_config.from_token = None
+
+ events, tokens = yield self.notifier.get_events_for(auth_user, pagin_config, timeout)
+
+ chunk = {
+ "chunk": [e.get_dict() for e in events],
+ "start_token": tokens[0].to_string(),
+ "end_token": tokens[1].to_string(),
+ }
+
+ defer.returnValue(chunk)
+
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index be10162db5..30d6269e2e 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -17,7 +17,6 @@ from twisted.internet import defer
from synapse.api.errors import SynapseError, AuthError
from synapse.api.constants import PresenceState
-from synapse.api.streams import StreamData
from ._base import BaseHandler
@@ -682,41 +681,10 @@ class PresenceHandler(BaseHandler):
user=observed_user,
clock=self.clock
),
- stream_type=PresenceStreamData,
store_id=statuscache.serial
)
-class PresenceStreamData(StreamData):
- def __init__(self, hs):
- super(PresenceStreamData, self).__init__(hs)
- self.presence = hs.get_handlers().presence_handler
-
- def get_rows(self, user_id, from_key, to_key, limit, direction):
- from_key = int(from_key)
- to_key = int(to_key)
-
- cachemap = self.presence._user_cachemap
-
- # TODO(paul): limit, and filter by visibility
- updates = [(k, cachemap[k]) for k in cachemap
- if from_key < cachemap[k].serial <= to_key]
-
- if updates:
- clock = self.presence.clock
-
- latest_serial = max([x[1].serial for x in updates])
- data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
- return ((data, latest_serial))
- else:
- return (([], self.presence._user_cachemap_latest_serial))
-
- def max_token(self):
- return self.presence._user_cachemap_latest_serial
-
-PresenceStreamData.EVENT_TYPE = PresenceStreamData
-
-
class UserPresenceCache(object):
"""Store an observed user's state and status message.
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 5a4569ac95..20b4bbb665 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -22,8 +22,6 @@ from synapse.api.errors import RoomError, StoreError, SynapseError
from synapse.api.events.room import (
RoomTopicEvent, RoomMemberEvent, RoomConfigEvent
)
-from synapse.api.streams.event import EventStream, EventsStreamData
-from synapse.handlers.presence import PresenceStreamData
from synapse.util import stringutils
from ._base import BaseHandler
@@ -115,13 +113,24 @@ class MessageHandler(BaseHandler):
"""
yield self.auth.check_joined_room(room_id, user_id)
- data_source = [
- EventsStreamData(self.hs, room_id=room_id, feedback=feedback)
- ]
- event_stream = EventStream(user_id, data_source)
- pagin_config = yield event_stream.fix_tokens(pagin_config)
- data_chunk = yield event_stream.get_chunk(config=pagin_config)
- defer.returnValue(data_chunk)
+ data_source = self.hs.get_event_sources().sources[0]
+
+ if pagin_config.from_token:
+ from_token = pagin_config.from_token
+ else:
+ from_token = yield self.hs.get_event_sources().get_current_token()
+
+ events, next_token = yield data_source.get_pagination_rows(
+ from_token, pagin_config.to_token, pagin_config.limit, room_id
+ )
+
+ chunk = {
+ "chunk": [e.get_dict() for e in events],
+ "start_token": from_token.to_string(),
+ "end_token": next_token.to_string(),
+ }
+
+ defer.returnValue(chunk)
@defer.inlineCallbacks
def store_room_data(self, event=None, stamp_event=True):
@@ -258,18 +267,15 @@ class MessageHandler(BaseHandler):
rooms_ret = []
- now_rooms_token = yield self.store.get_room_events_max_id()
+ # FIXME (erikj): We need to not generate this token,
+ now_token = yield self.hs.get_event_sources().get_current_token()
# FIXME (erikj): Fix this.
- presence_stream = PresenceStreamData(self.hs)
- now_presence_token = yield presence_stream.max_token()
- presence = yield presence_stream.get_rows(
- user_id, 0, now_presence_token, None, None
+ presence_stream = self.hs.get_event_sources().sources[1]
+ presence = yield presence_stream.get_new_events_for_user(
+ user_id, now_token, None, None
)
- # FIXME (erikj): We need to not generate this token,
- now_token = "%s_%s" % (now_rooms_token, now_presence_token)
-
limit = pagin_config.limit
if not limit:
limit = 10
@@ -291,7 +297,7 @@ class MessageHandler(BaseHandler):
messages, token = yield self.store.get_recent_events_for_room(
event.room_id,
limit=limit,
- end_token=now_rooms_token,
+ end_token=now_token.events_key.to_string(),
)
d["messages"] = {
@@ -305,9 +311,7 @@ class MessageHandler(BaseHandler):
except:
logger.exception("Failed to get snapshot")
- ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token}
-
- # logger.debug("snapshot_all_rooms returning: %s", ret)
+ ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token.to_string()}
defer.returnValue(ret)
|
