def get_recent_event_ids_for_room(self, room_id, limit, end_token, from_token=None):
        end_token = RoomStreamToken.parse_stream_token(end_token)

        if from_token is None:
            sql = (
                "SELECT stream_ordering, topological_ordering, event_id"
                " FROM events"
                " WHERE room_id = ? AND stream_ordering <= ? AND outlier = ?"
                " ORDER BY topological_ordering DESC, stream_ordering DESC"
                " LIMIT ?"
            )
        else:
            from_token = RoomStreamToken.parse_stream_token(from_token)
            sql = (
                "SELECT stream_ordering, topological_ordering, event_id"
                " FROM events"
                " WHERE room_id = ? AND stream_ordering > ?"
                " AND stream_ordering <= ? AND outlier = ?"
                " ORDER BY topological_ordering DESC, stream_ordering DESC"
                " LIMIT ?"
            )

        def get_recent_events_for_room_txn(txn):
            if from_token is None:
                txn.execute(sql, (room_id, end_token.stream, False, limit,))
            else:
                txn.execute(sql, (
                    room_id, from_token.stream, end_token.stream, False, limit
                ))

            rows = self.cursor_to_dict(txn)

            rows.reverse()  # As we selected with reverse ordering

            if rows:
                # Tokens are positions between events.
                # This token points *after* the last event in the chunk.
                # We need it to point to the event before it in the chunk
                # since we are going backwards so we subtract one from the
                # stream part.
                topo = rows[0]["topological_ordering"]
                toke = rows[0]["stream_ordering"] - 1
                start_token = str(RoomStreamToken(topo, toke))

                token = (start_token, str(end_token))
            else:
                token = (str(end_token), str(end_token))

            return rows, token

        return self.runInteraction(
            "get_recent_events_for_room", get_recent_events_for_room_txn
        )

    def get_membership_changes_for_user(self, user_id, from_key, to_key):
        if from_key is not None:
            from_id = RoomStreamToken.parse_stream_token(from_key).stream
        else:
            from_id = None
        to_id = RoomStreamToken.parse_stream_token(to_key).stream

        if from_key == to_key:
            defer.returnValue([])

        if from_id:
            has_changed = self._membership_stream_cache.has_entity_changed(
                user_id, int(from_id)
            )
            if not has_changed:
                defer.returnValue([])

        def f(txn):
            if from_id is not None:
                sql = (
                    "SELECT m.event_id, stream_ordering FROM events AS e,"
                    " room_memberships AS m"
                    " WHERE e.event_id = m.event_id"
                    " AND m.user_id = ?"
                    " AND e.stream_ordering > ? AND e.stream_ordering <= ?"
                    " ORDER BY e.stream_ordering ASC"
                )
                txn.execute(sql, (user_id, from_id, to_id,))
            else:
                sql = (
                    "SELECT m.event_id, stream_ordering FROM events AS e,"
                    " room_memberships AS m"
                    " WHERE e.event_id = m.event_id"
                    " AND m.user_id = ?"
                    " AND stream_ordering <= ?"
                    " ORDER BY stream_ordering ASC"
                )
                txn.execute(sql, (user_id, to_id,))
            rows = self.cursor_to_dict(txn)

            return rows

        rows = yield self.runInteraction("get_membership_changes_for_user", f)

        ret = yield self._get_events(
            [r["event_id"] for r in rows],
            get_prev_content=True
        )

        self._set_before_and_after(ret, rows, topo_order=False)

        defer.returnValue(ret)

    def get_recent_event_ids_for_room(self, room_id, limit, end_token):
        """Get the most recent events in the room in topological ordering.

        Args:
            room_id (str)
            limit (int)
            end_token (str): The stream token representing now.

        Returns:
            Deferred[tuple[list[_EventDictReturn], str]]: Returns a list of
            _EventDictReturn and a token pointing to the start of the returned
            events.
            The events returned are in ascending order.
        """
        # Allow a zero limit here, and no-op.
        if limit == 0:
            defer.returnValue(([], end_token))

        end_token = RoomStreamToken.parse(end_token)

        rows, token = yield self.runInteraction(
            "get_recent_event_ids_for_room", self._paginate_room_events_txn,
            room_id, from_token=end_token, limit=limit,
        )

        # We want to return the results in ascending order.
        rows.reverse()

        defer.returnValue((rows, token))

    def get_messages(self, user_id=None, room_id=None, pagin_config=None,
                     feedback=False, as_client_event=True):
        """Get messages in a room.

        Args:
            user_id (str): The user requesting messages.
            room_id (str): The room they want messages from.
            pagin_config (synapse.api.streams.PaginationConfig): The pagination
            config rules to apply, if any.
            feedback (bool): True to get compressed feedback with the messages
            as_client_event (bool): True to get events in client-server format.
        Returns:
            dict: Pagination API results
        """
        yield self.auth.check_joined_room(room_id, user_id)

        data_source = self.hs.get_event_sources().sources["room"]

        if not pagin_config.from_token:
            pagin_config.from_token = (
                yield self.hs.get_event_sources().get_current_token(
                    direction='b'
                )
            )

        room_token = RoomStreamToken.parse(pagin_config.from_token.room_key)
        if room_token.topological is None:
            raise SynapseError(400, "Invalid token")

        yield self.hs.get_handlers().federation_handler.maybe_backfill(
            room_id, room_token.topological
        )

        user = UserID.from_string(user_id)

        events, next_key = yield data_source.get_pagination_rows(
            user, pagin_config.get_source_config("room"), room_id
        )

        next_token = pagin_config.from_token.copy_and_replace(
            "room_key", next_key
        )

        time_now = self.clock.time_msec()

        chunk = {
            "chunk": [
                serialize_event(e, time_now, as_client_event) for e in events
            ],
            "start": pagin_config.from_token.to_string(),
            "end": next_token.to_string(),
        }

        defer.returnValue(chunk)

    def get_new_events(
            self,
            user,
            from_key,
            limit,
            room_ids,
            is_guest,
    ):
        # We just ignore the key for now.

        to_key = yield self.get_current_key()

        from_token = RoomStreamToken.parse(from_key)
        if from_token.topological:
            logger.warn("Stream has topological part!!!! %r", from_key)
            from_key = "s%s" % (from_token.stream,)

        app_service = yield self.store.get_app_service_by_user_id(
            user.to_string()
        )
        if app_service:
            events, end_key = yield self.store.get_appservice_room_stream(
                service=app_service,
                from_key=from_key,
                to_key=to_key,
                limit=limit,
            )
        else:
            room_events = yield self.store.get_membership_changes_for_user(
                user.to_string(), from_key, to_key
            )

            room_to_events = yield self.store.get_room_events_stream_for_rooms(
                room_ids=room_ids,
                from_key=from_key,
                to_key=to_key,
                limit=limit or 10,
                order='ASC',
            )

            events = list(room_events)
            events.extend(e for evs, _ in room_to_events.values() for e in evs)

            events.sort(key=lambda e: e.internal_metadata.order)

            if limit:
                events[:] = events[:limit]

            if events:
                end_key = events[-1].internal_metadata.after
            else:
                end_key = to_key

        defer.returnValue((events, end_key))

    def get_new_events(
            self,
            user,
            from_key,
            limit,
            room_ids,
            is_guest,
            explicit_room_id=None,
    ):
        # We just ignore the key for now.

        to_key = yield self.get_current_key()

        from_token = RoomStreamToken.parse(from_key)
        if from_token.topological:
            logger.warn("Stream has topological part!!!! %r", from_key)
            from_key = "s%s" % (from_token.stream,)

        app_service = self.store.get_app_service_by_user_id(
            user.to_string()
        )
        if app_service:
            # We no longer support AS users using /sync directly.
            # See https://github.com/matrix-org/matrix-doc/issues/1144
            raise NotImplementedError()
        else:
            room_events = yield self.store.get_membership_changes_for_user(
                user.to_string(), from_key, to_key
            )

            room_to_events = yield self.store.get_room_events_stream_for_rooms(
                room_ids=room_ids,
                from_key=from_key,
                to_key=to_key,
                limit=limit or 10,
                order='ASC',
            )

            events = list(room_events)
            events.extend(e for evs, _ in room_to_events.values() for e in evs)

            events.sort(key=lambda e: e.internal_metadata.order)

            if limit:
                events[:] = events[:limit]

            if events:
                end_key = events[-1].internal_metadata.after
            else:
                end_key = to_key

        defer.returnValue((events, end_key))

    def get_rooms_that_changed(self, room_ids, from_key):
        """Given a list of rooms and a token, return rooms where there may have
        been changes.

        Args:
            room_ids (list)
            from_key (str): The room_key portion of a StreamToken
        """
        from_key = RoomStreamToken.parse_stream_token(from_key).stream
        return set(
            room_id for room_id in room_ids
            if self._events_stream_cache.has_entity_changed(room_id, from_key)
        )

    def paginate_room_events(self, room_id, from_key, to_key=None,
                             direction='b', limit=-1, event_filter=None):
        """Returns list of events before or after a given token.

        Args:
            room_id (str)
            from_key (str): The token used to stream from
            to_key (str|None): A token which if given limits the results to
                only those before
            direction(char): Either 'b' or 'f' to indicate whether we are
                paginating forwards or backwards from `from_key`.
            limit (int): The maximum number of events to return. Zero or less
                means no limit.
            event_filter (Filter|None): If provided filters the events to
                those that match the filter.

        Returns:
            tuple[list[dict], str]: Returns the results as a list of dicts and
            a token that points to the end of the result set. The dicts have
            the keys "event_id", "topological_ordering" and "stream_orderign".
        """

        from_key = RoomStreamToken.parse(from_key)
        if to_key:
            to_key = RoomStreamToken.parse(to_key)

        rows, token = yield self.runInteraction(
            "paginate_room_events", self._paginate_room_events_txn,
            room_id, from_key, to_key, direction, limit, event_filter,
        )

        events = yield self._get_events(
            [r.event_id for r in rows],
            get_prev_content=True
        )

        self._set_before_and_after(events, rows)

        defer.returnValue((events, token))

    def get_room_events_stream_for_rooms(self, room_ids, from_key, to_key, limit=0,
                                         order='DESC'):
        from_id = RoomStreamToken.parse_stream_token(from_key).stream

        room_ids = yield self._events_stream_cache.get_entities_changed(
            room_ids, from_id
        )

        if not room_ids:
            defer.returnValue({})

        results = {}
        room_ids = list(room_ids)
        for rm_ids in (room_ids[i:i + 20] for i in xrange(0, len(room_ids), 20)):
            res = yield defer.gatherResults([
                preserve_fn(self.get_room_events_stream_for_room)(
                    room_id, from_key, to_key, limit, order=order,
                )
                for room_id in rm_ids
            ])
            results.update(dict(zip(rm_ids, res)))

        defer.returnValue(results)

    def get_room_events_stream_for_rooms(self, room_ids, from_key, to_key, limit=0,
                                         order='DESC'):
        from_id = RoomStreamToken.parse_stream_token(from_key).stream

        room_ids = yield self._events_stream_cache.get_entities_changed(
            room_ids, from_id
        )

        if not room_ids:
            defer.returnValue({})

        results = {}
        room_ids = list(room_ids)
        for rm_ids in (room_ids[i:i + 20] for i in range(0, len(room_ids), 20)):
            res = yield make_deferred_yieldable(defer.gatherResults([
                run_in_background(
                    self.get_room_events_stream_for_room,
                    room_id, from_key, to_key, limit, order=order,
                )
                for room_id in rm_ids
            ], consumeErrors=True))
            results.update(dict(zip(rm_ids, res)))

        defer.returnValue(results)

    def get_room_events_stream_for_room(self, room_id, from_key, to_key, limit=0,
                                        order='DESC'):
        # Note: If from_key is None then we return in topological order. This
        # is because in that case we're using this as a "get the last few messages
        # in a room" function, rather than "get new messages since last sync"
        if from_key is not None:
            from_id = RoomStreamToken.parse_stream_token(from_key).stream
        else:
            from_id = None
        to_id = RoomStreamToken.parse_stream_token(to_key).stream

        if from_key == to_key:
            defer.returnValue(([], from_key))

        if from_id:
            has_changed = yield self._events_stream_cache.has_entity_changed(
                room_id, from_id
            )

            if not has_changed:
                defer.returnValue(([], from_key))

        def f(txn):
            if from_id is not None:
                sql = (
                    "SELECT event_id, stream_ordering FROM events WHERE"
                    " room_id = ?"
                    " AND not outlier"
                    " AND stream_ordering > ? AND stream_ordering <= ?"
                    " ORDER BY stream_ordering %s LIMIT ?"
                ) % (order,)
                txn.execute(sql, (room_id, from_id, to_id, limit))
            else:
                sql = (
                    "SELECT event_id, stream_ordering FROM events WHERE"
                    " room_id = ?"
                    " AND not outlier"
                    " AND stream_ordering <= ?"
                    " ORDER BY topological_ordering %s, stream_ordering %s LIMIT ?"
                ) % (order, order,)
                txn.execute(sql, (room_id, to_id, limit))

            rows = self.cursor_to_dict(txn)

            return rows

        rows = yield self.runInteraction("get_room_events_stream_for_room", f)

        ret = yield self._get_events(
            [r["event_id"] for r in rows],
            get_prev_content=True
        )

        self._set_before_and_after(ret, rows, topo_order=from_id is None)

        if order.lower() == "desc":
            ret.reverse()

        if rows:
            key = "s%d" % min(r["stream_ordering"] for r in rows)
        else:
            # Assume we didn't get anything because there was nothing to
            # get.
            key = from_key

        defer.returnValue((ret, key))

    def get_appservice_room_stream(self, service, from_key, to_key, limit=0):
        # NB this lives here instead of appservice.py so we can reuse the
        # 'private' StreamToken class in this file.
        if limit:
            limit = max(limit, MAX_STREAM_SIZE)
        else:
            limit = MAX_STREAM_SIZE

        # From and to keys should be integers from ordering.
        from_id = RoomStreamToken.parse_stream_token(from_key)
        to_id = RoomStreamToken.parse_stream_token(to_key)

        if from_key == to_key:
            defer.returnValue(([], to_key))
            return

        # select all the events between from/to with a sensible limit
        sql = (
            "SELECT e.event_id, e.room_id, e.type, s.state_key, "
            "e.stream_ordering FROM events AS e "
            "LEFT JOIN state_events as s ON "
            "e.event_id = s.event_id "
            "WHERE e.stream_ordering > ? AND e.stream_ordering <= ? "
            "ORDER BY stream_ordering ASC LIMIT %(limit)d "
        ) % {
            "limit": limit
        }

        def f(txn):
            # pull out all the events between the tokens
            txn.execute(sql, (from_id.stream, to_id.stream,))
            rows = self.cursor_to_dict(txn)

            # Logic:
            #  - We want ALL events which match the AS room_id regex
            #  - We want ALL events which match the rooms represented by the AS
            #    room_alias regex
            #  - We want ALL events for rooms that AS users have joined.
            # This is currently supported via get_app_service_rooms (which is
            # used for the Notifier listener rooms). We can't reasonably make a
            # SQL query for these room IDs, so we'll pull all the events between
            # from/to and filter in python.
            rooms_for_as = self._get_app_service_rooms_txn(txn, service)
            room_ids_for_as = [r.room_id for r in rooms_for_as]

            def app_service_interested(row):
                if row["room_id"] in room_ids_for_as:
                    return True

                if row["type"] == EventTypes.Member:
                    if service.is_interested_in_user(row.get("state_key")):
                        return True
                return False

            return [r for r in rows if app_service_interested(r)]

        rows = yield self.runInteraction("get_appservice_room_stream", f)

        ret = yield self._get_events(
            [r["event_id"] for r in rows],
            get_prev_content=True
        )

        self._set_before_and_after(ret, rows, topo_order=from_id is None)

        if rows:
            key = "s%d" % max(r["stream_ordering"] for r in rows)
        else:
            # Assume we didn't get anything because there was nothing to
            # get.
            key = to_key

        defer.returnValue((ret, key))

    def get_messages(self, requester, room_id=None, pagin_config=None,
                     as_client_event=True, event_filter=None):
        """Get messages in a room.

        Args:
            requester (Requester): The user requesting messages.
            room_id (str): The room they want messages from.
            pagin_config (synapse.api.streams.PaginationConfig): The pagination
                config rules to apply, if any.
            as_client_event (bool): True to get events in client-server format.
            event_filter (Filter): Filter to apply to results or None
        Returns:
            dict: Pagination API results
        """
        user_id = requester.user.to_string()

        if pagin_config.from_token:
            room_token = pagin_config.from_token.room_key
        else:
            pagin_config.from_token = (
                yield self.hs.get_event_sources().get_current_token_for_room(
                    room_id=room_id
                )
            )
            room_token = pagin_config.from_token.room_key

        room_token = RoomStreamToken.parse(room_token)

        pagin_config.from_token = pagin_config.from_token.copy_and_replace(
            "room_key", str(room_token)
        )

        source_config = pagin_config.get_source_config("room")

        with (yield self.pagination_lock.read(room_id)):
            membership, member_event_id = yield self.auth.check_in_room_or_world_readable(
                room_id, user_id
            )

            if source_config.direction == 'b':
                # if we're going backwards, we might need to backfill. This
                # requires that we have a topo token.
                if room_token.topological:
                    max_topo = room_token.topological
                else:
                    max_topo = yield self.store.get_max_topological_token(
                        room_id, room_token.stream
                    )

                if membership == Membership.LEAVE:
                    # If they have left the room then clamp the token to be before
                    # they left the room, to save the effort of loading from the
                    # database.
                    leave_token = yield self.store.get_topological_token_for_event(
                        member_event_id
                    )
                    leave_token = RoomStreamToken.parse(leave_token)
                    if leave_token.topological < max_topo:
                        source_config.from_key = str(leave_token)

                yield self.hs.get_handlers().federation_handler.maybe_backfill(
                    room_id, max_topo
                )

            events, next_key = yield self.store.paginate_room_events(
                room_id=room_id,
                from_key=source_config.from_key,
                to_key=source_config.to_key,
                direction=source_config.direction,
                limit=source_config.limit,
                event_filter=event_filter,
            )

            next_token = pagin_config.from_token.copy_and_replace(
                "room_key", next_key
            )

        if events:
            if event_filter:
                events = event_filter.filter(events)

            events = yield filter_events_for_client(
                self.store,
                user_id,
                events,
                is_peeking=(member_event_id is None),
            )

        if not events:
            defer.returnValue({
                "chunk": [],
                "start": pagin_config.from_token.to_string(),
                "end": next_token.to_string(),
            })

        state = None
        if event_filter and event_filter.lazy_load_members():
            # TODO: remove redundant members

            # FIXME: we also care about invite targets etc.
#.........这里部分代码省略.........

    def get_messages(self, requester, room_id=None, pagin_config=None,
                     as_client_event=True):
        """Get messages in a room.

        Args:
            requester (Requester): The user requesting messages.
            room_id (str): The room they want messages from.
            pagin_config (synapse.api.streams.PaginationConfig): The pagination
                config rules to apply, if any.
            as_client_event (bool): True to get events in client-server format.
        Returns:
            dict: Pagination API results
        """
        user_id = requester.user.to_string()
        data_source = self.hs.get_event_sources().sources["room"]

        if pagin_config.from_token:
            room_token = pagin_config.from_token.room_key
        else:
            pagin_config.from_token = (
                yield self.hs.get_event_sources().get_current_token(
                    direction='b'
                )
            )
            room_token = pagin_config.from_token.room_key

        room_token = RoomStreamToken.parse(room_token)

        pagin_config.from_token = pagin_config.from_token.copy_and_replace(
            "room_key", str(room_token)
        )

        source_config = pagin_config.get_source_config("room")

        membership, member_event_id = yield self._check_in_room_or_world_readable(
            room_id, user_id
        )

        if source_config.direction == 'b':
            # if we're going backwards, we might need to backfill. This
            # requires that we have a topo token.
            if room_token.topological:
                max_topo = room_token.topological
            else:
                max_topo = yield self.store.get_max_topological_token_for_stream_and_room(
                    room_id, room_token.stream
                )

            if membership == Membership.LEAVE:
                # If they have left the room then clamp the token to be before
                # they left the room, to save the effort of loading from the
                # database.
                leave_token = yield self.store.get_topological_token_for_event(
                    member_event_id
                )
                leave_token = RoomStreamToken.parse(leave_token)
                if leave_token.topological < max_topo:
                    source_config.from_key = str(leave_token)

            yield self.hs.get_handlers().federation_handler.maybe_backfill(
                room_id, max_topo
            )

        events, next_key = yield data_source.get_pagination_rows(
            requester.user, source_config, room_id
        )

        next_token = pagin_config.from_token.copy_and_replace(
            "room_key", next_key
        )

        if not events:
            defer.returnValue({
                "chunk": [],
                "start": pagin_config.from_token.to_string(),
                "end": next_token.to_string(),
            })

        events = yield self._filter_events_for_client(
            user_id,
            events,
            is_peeking=(member_event_id is None),
        )

        time_now = self.clock.time_msec()

        chunk = {
            "chunk": [
                serialize_event(e, time_now, as_client_event)
                for e in events
            ],
            "start": pagin_config.from_token.to_string(),
            "end": next_token.to_string(),
        }

        defer.returnValue(chunk)

    def get_room_events_stream(
        self,
        user_id,
        from_key,
        to_key,
        limit=0,
        is_guest=False,
        room_ids=None
    ):
        room_ids = room_ids or []
        room_ids = [r for r in room_ids]
        if is_guest:
            current_room_membership_sql = (
                "SELECT c.room_id FROM history_visibility AS h"
                " INNER JOIN current_state_events AS c"
                " ON h.event_id = c.event_id"
                " WHERE c.room_id IN (%s)"
                " AND h.history_visibility = 'world_readable'" % (
                    ",".join(map(lambda _: "?", room_ids))
                )
            )
            current_room_membership_args = room_ids
        else:
            current_room_membership_sql = (
                "SELECT m.room_id FROM room_memberships as m "
                " INNER JOIN current_state_events as c"
                " ON m.event_id = c.event_id AND c.state_key = m.user_id"
                " WHERE m.user_id = ? AND m.membership = 'join'"
            )
            current_room_membership_args = [user_id]

        # We also want to get any membership events about that user, e.g.
        # invites or leave notifications.
        membership_sql = (
            "SELECT m.event_id FROM room_memberships as m "
            "INNER JOIN current_state_events as c ON m.event_id = c.event_id "
            "WHERE m.user_id = ? "
        )
        membership_args = [user_id]

        if limit:
            limit = max(limit, MAX_STREAM_SIZE)
        else:
            limit = MAX_STREAM_SIZE

        # From and to keys should be integers from ordering.
        from_id = RoomStreamToken.parse_stream_token(from_key)
        to_id = RoomStreamToken.parse_stream_token(to_key)

        if from_key == to_key:
            return defer.succeed(([], to_key))

        sql = (
            "SELECT e.event_id, e.stream_ordering FROM events AS e WHERE "
            "(e.outlier = ? AND (room_id IN (%(current)s)) OR "
            "(event_id IN (%(invites)s))) "
            "AND e.stream_ordering > ? AND e.stream_ordering <= ? "
            "ORDER BY stream_ordering ASC LIMIT %(limit)d "
        ) % {
            "current": current_room_membership_sql,
            "invites": membership_sql,
            "limit": limit
        }

        def f(txn):
            args = ([False] + current_room_membership_args + membership_args +
                    [from_id.stream, to_id.stream])
            txn.execute(sql, args)

            rows = self.cursor_to_dict(txn)

            ret = self._get_events_txn(
                txn,
                [r["event_id"] for r in rows],
                get_prev_content=True
            )

            self._set_before_and_after(ret, rows)

            if rows:
                key = "s%d" % max(r["stream_ordering"] for r in rows)
            else:
                # Assume we didn't get anything because there was nothing to
                # get.
                key = to_key

            return ret, key

        return self.runInteraction("get_room_events_stream", f)

    def get_recent_events_for_room(self, room_id, limit, end_token, from_token=None):
        # TODO (erikj): Handle compressed feedback

        end_token = RoomStreamToken.parse_stream_token(end_token)

        if from_token is None:
            sql = (
                "SELECT stream_ordering, topological_ordering, event_id"
                " FROM events"
                " WHERE room_id = ? AND stream_ordering <= ? AND outlier = ?"
                " ORDER BY topological_ordering DESC, stream_ordering DESC"
                " LIMIT ?"
            )
        else:
            from_token = RoomStreamToken.parse_stream_token(from_token)
            sql = (
                "SELECT stream_ordering, topological_ordering, event_id"
                " FROM events"
                " WHERE room_id = ? AND stream_ordering > ?"
                " AND stream_ordering <= ? AND outlier = ?"
                " ORDER BY topological_ordering DESC, stream_ordering DESC"
                " LIMIT ?"
            )

        def get_recent_events_for_room_txn(txn):
            if from_token is None:
                txn.execute(sql, (room_id, end_token.stream, False, limit,))
            else:
                txn.execute(sql, (
                    room_id, from_token.stream, end_token.stream, False, limit
                ))

            rows = self.cursor_to_dict(txn)

            rows.reverse()  # As we selected with reverse ordering

            if rows:
                # Tokens are positions between events.
                # This token points *after* the last event in the chunk.
                # We need it to point to the event before it in the chunk
                # since we are going backwards so we subtract one from the
                # stream part.
                topo = rows[0]["topological_ordering"]
                toke = rows[0]["stream_ordering"] - 1
                start_token = str(RoomStreamToken(topo, toke))

                token = (start_token, str(end_token))
            else:
                token = (str(end_token), str(end_token))

            return rows, token

        rows, token = yield self.runInteraction(
            "get_recent_events_for_room", get_recent_events_for_room_txn
        )

        logger.debug("stream before")
        events = yield self._get_events(
            [r["event_id"] for r in rows],
            get_prev_content=True
        )
        logger.debug("stream after")

        self._set_before_and_after(events, rows)

        defer.returnValue((events, token))

    def paginate_room_events(self, room_id, from_key, to_key=None,
                             direction='b', limit=-1, event_filter=None):
        # Tokens really represent positions between elements, but we use
        # the convention of pointing to the event before the gap. Hence
        # we have a bit of asymmetry when it comes to equalities.
        args = [False, room_id]
        if direction == 'b':
            order = "DESC"
            bounds = upper_bound(
                RoomStreamToken.parse(from_key), self.database_engine
            )
            if to_key:
                bounds = "%s AND %s" % (bounds, lower_bound(
                    RoomStreamToken.parse(to_key), self.database_engine
                ))
        else:
            order = "ASC"
            bounds = lower_bound(
                RoomStreamToken.parse(from_key), self.database_engine
            )
            if to_key:
                bounds = "%s AND %s" % (bounds, upper_bound(
                    RoomStreamToken.parse(to_key), self.database_engine
                ))

        filter_clause, filter_args = filter_to_clause(event_filter)

        if filter_clause:
            bounds += " AND " + filter_clause
            args.extend(filter_args)

        if int(limit) > 0:
            args.append(int(limit))
            limit_str = " LIMIT ?"
        else:
            limit_str = ""

        sql = (
            "SELECT * FROM events"
            " WHERE outlier = ? AND room_id = ? AND %(bounds)s"
            " ORDER BY topological_ordering %(order)s,"
            " stream_ordering %(order)s %(limit)s"
        ) % {
            "bounds": bounds,
            "order": order,
            "limit": limit_str
        }

        def f(txn):
            txn.execute(sql, args)

            rows = self.cursor_to_dict(txn)

            if rows:
                topo = rows[-1]["topological_ordering"]
                toke = rows[-1]["stream_ordering"]
                if direction == 'b':
                    # Tokens are positions between events.
                    # This token points *after* the last event in the chunk.
                    # We need it to point to the event before it in the chunk
                    # when we are going backwards so we subtract one from the
                    # stream part.
                    toke -= 1
                next_token = str(RoomStreamToken(topo, toke))
            else:
                # TODO (erikj): We should work out what to do here instead.
                next_token = to_key if to_key else from_key

            return rows, next_token,

        rows, token = yield self.runInteraction("paginate_room_events", f)

        events = yield self._get_events(
            [r["event_id"] for r in rows],
            get_prev_content=True
        )

        self._set_before_and_after(events, rows)

        defer.returnValue((events, token))

    def get_messages(self, user_id=None, room_id=None, pagin_config=None,
                     as_client_event=True, is_guest=False):
        """Get messages in a room.

        Args:
            user_id (str): The user requesting messages.
            room_id (str): The room they want messages from.
            pagin_config (synapse.api.streams.PaginationConfig): The pagination
                config rules to apply, if any.
            as_client_event (bool): True to get events in client-server format.
            is_guest (bool): Whether the requesting user is a guest (as opposed
                to a fully registered user).
        Returns:
            dict: Pagination API results
        """
        data_source = self.hs.get_event_sources().sources["room"]

        if pagin_config.from_token:
            room_token = pagin_config.from_token.room_key
        else:
            pagin_config.from_token = (
                yield self.hs.get_event_sources().get_current_token(
                    direction='b'
                )
            )
            room_token = pagin_config.from_token.room_key

        room_token = RoomStreamToken.parse(room_token)
        if room_token.topological is None:
            raise SynapseError(400, "Invalid token")

        pagin_config.from_token = pagin_config.from_token.copy_and_replace(
            "room_key", str(room_token)
        )

        source_config = pagin_config.get_source_config("room")

        if not is_guest:
            member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
            if member_event.membership == Membership.LEAVE:
                # If they have left the room then clamp the token to be before
                # they left the room.
                # If they're a guest, we'll just 403 them if they're asking for
                # events they can't see.
                leave_token = yield self.store.get_topological_token_for_event(
                    member_event.event_id
                )
                leave_token = RoomStreamToken.parse(leave_token)
                if leave_token.topological < room_token.topological:
                    source_config.from_key = str(leave_token)