def on_GET(self, request, room_id, event_id):
        requester = yield self.auth.get_user_by_req(request, allow_guest=True)

        limit = int(request.args.get("limit", [10])[0])

        results = yield self.handlers.room_context_handler.get_event_context(
            requester.user,
            room_id,
            event_id,
            limit,
            requester.is_guest,
        )

        if not results:
            raise SynapseError(
                404, "Event not found.", errcode=Codes.NOT_FOUND
            )

        time_now = self.clock.time_msec()
        results["events_before"] = [
            serialize_event(event, time_now) for event in results["events_before"]
        ]
        results["event"] = serialize_event(results["event"], time_now)
        results["events_after"] = [
            serialize_event(event, time_now) for event in results["events_after"]
        ]
        results["state"] = [
            serialize_event(event, time_now) for event in results["state"]
        ]

        defer.returnValue((200, results))

    def encode_room(room, filter, time_now, token_id):
        event_map = {}
        state_events = filter.filter_room_state(room.state)
        recent_events = filter.filter_room_events(room.events)
        state_event_ids = []
        recent_event_ids = []
        for event in state_events:
            # TODO(mjark): Respect formatting requirements in the filter.
            event_map[event.event_id] = serialize_event(
                event, time_now, token_id=token_id,
                event_format=format_event_for_client_v2_without_event_id,
            )
            state_event_ids.append(event.event_id)

        for event in recent_events:
            # TODO(mjark): Respect formatting requirements in the filter.
            event_map[event.event_id] = serialize_event(
                event, time_now, token_id=token_id,
                event_format=format_event_for_client_v2_without_event_id,
            )
            recent_event_ids.append(event.event_id)
        result = {
            "room_id": room.room_id,
            "event_map": event_map,
            "events": {
                "batch": recent_event_ids,
                "prev_batch": room.prev_batch.to_string(),
            },
            "state": state_event_ids,
            "limited": room.limited,
            "published": room.published,
            "ephemeral": room.ephemeral,
        }
        return result

        def handle_room(event):
            d = {
                "room_id": event.room_id,
                "membership": event.membership,
                "visibility": (
                    "public" if event.room_id in public_room_ids
                    else "private"
                ),
            }

            if event.membership == Membership.INVITE:
                d["inviter"] = event.sender

            rooms_ret.append(d)

            if event.membership != Membership.JOIN:
                return
            try:
                (messages, token), current_state = yield defer.gatherResults(
                    [
                        self.store.get_recent_events_for_room(
                            event.room_id,
                            limit=limit,
                            end_token=now_token.room_key,
                        ),
                        self.state_handler.get_current_state(
                            event.room_id
                        ),
                    ]
                ).addErrback(unwrapFirstError)

                messages = yield self._filter_events_for_client(
                    user_id, event.room_id, messages
                )

                start_token = now_token.copy_and_replace("room_key", token[0])
                end_token = now_token.copy_and_replace("room_key", token[1])
                time_now = self.clock.time_msec()

                d["messages"] = {
                    "chunk": [
                        serialize_event(m, time_now, as_client_event)
                        for m in messages
                    ],
                    "start": start_token.to_string(),
                    "end": end_token.to_string(),
                }

                d["state"] = [
                    serialize_event(c, time_now, as_client_event)
                    for c in current_state.values()
                ]
            except:
                logger.exception("Failed to get snapshot")

    def get_room_members_as_pagination_chunk(self, room_id=None, user_id=None,
                                             limit=0, start_tok=None,
                                             end_tok=None):
        """Retrieve a list of room members in the room.

        Args:
            room_id (str): The room to get the member list for.
            user_id (str): The ID of the user making the request.
            limit (int): The max number of members to return.
            start_tok (str): Optional. The start token if known.
            end_tok (str): Optional. The end token if known.
        Returns:
            dict: A Pagination streamable dict.
        Raises:
            SynapseError if something goes wrong.
        """
        yield self.auth.check_joined_room(room_id, user_id)

        member_list = yield self.store.get_room_members(room_id=room_id)
        time_now = self.clock.time_msec()
        event_list = [
            serialize_event(entry, time_now)
            for entry in member_list
        ]
        chunk_data = {
            "start": "START",  # FIXME (erikj): START is no longer valid
            "end": "END",
            "chunk": event_list
        }
        # TODO honor Pagination stream params
        # TODO snapshot this list to return on subsequent requests when
        # paginating
        defer.returnValue(chunk_data)

 def serialize(event):
     # TODO(mjark): Respect formatting requirements in the filter.
     return serialize_event(
         event, time_now, token_id=token_id,
         event_format=format_event_for_client_v2_without_room_id,
         only_event_fields=only_fields,
     )

    def encode_invited(rooms, time_now, token_id):
        """
        Encode the invited rooms in a sync result

        Args:
            rooms(list[synapse.handlers.sync.InvitedSyncResult]): list of
                sync results for rooms this user is joined to
            time_now(int): current time - used as a baseline for age
                calculations
            token_id(int): ID of the user's auth token - used for namespacing
            of transaction IDs

        Returns:
            dict[str, dict[str, object]]: the invited rooms list, in our
                response format
        """
        invited = {}
        for room in rooms:
            invite = serialize_event(
                room.invite, time_now, token_id=token_id,
                event_format=format_event_for_client_v2_without_room_id,
                is_invite=True,
            )
            unsigned = dict(invite.get("unsigned", {}))
            invite["unsigned"] = unsigned
            invited_state = list(unsigned.pop("invite_room_state", []))
            invited_state.append(invite)
            invited[room.room_id] = {
                "invite_state": {"events": invited_state}
            }

        return invited

    def get_state_events(self, user_id, room_id, is_guest=False):
        """Retrieve all state events for a given room. If the user is
        joined to the room then return the current state. If the user has
        left the room return the state events from when they left.

        Args:
            user_id(str): The user requesting state events.
            room_id(str): The room ID to get all state events from.
        Returns:
            A list of dicts representing state events. [{}, {}, {}]
        """
        membership, membership_event_id = yield self._check_in_room_or_world_readable(
            room_id, user_id
        )

        if membership == Membership.JOIN:
            room_state = yield self.state_handler.get_current_state(room_id)
        elif membership == Membership.LEAVE:
            room_state = yield self.store.get_state_for_events(
                [membership_event_id], None
            )
            room_state = room_state[membership_event_id]

        now = self.clock.time_msec()
        defer.returnValue(
            [serialize_event(c, now) for c in room_state.values()]
        )

    def encode_invited(self, rooms, filter, time_now, token_id):
        """
        Encode the invited rooms in a sync result

        :param list[synapse.handlers.sync.InvitedSyncResult] rooms: list of
             sync results for rooms this user is joined to
        :param FilterCollection filter: filters to apply to the results
        :param int time_now: current time - used as a baseline for age
            calculations
        :param int token_id: ID of the user's auth token - used for namespacing
            of transaction IDs

        :return: the invited rooms list, in our response format
        :rtype: dict[str, dict[str, object]]
        """
        invited = {}
        for room in rooms:
            invite = serialize_event(
                room.invite, time_now, token_id=token_id,
                event_format=format_event_for_client_v2_without_room_id,
            )
            invited_state = invite.get("unsigned", {}).pop("invite_room_state", [])
            invited_state.append(invite)
            invited[room.room_id] = {
                "invite_state": {"events": invited_state}
            }

        return invited

    def on_GET(self, request, room_id, event_id):
        requester = yield self.auth.get_user_by_req(request, allow_guest=True)
        event = yield self.event_handler.get_event(requester.user, room_id, event_id)

        time_now = self.clock.time_msec()
        if event:
            defer.returnValue((200, serialize_event(event, time_now)))
        else:
            defer.returnValue((404, "Event not found."))

    def _room_initial_sync_parted(self, user_id, room_id, pagin_config,
                                  membership, member_event_id, is_peeking):
        room_state = yield self.store.get_state_for_events(
            [member_event_id], None
        )

        room_state = room_state[member_event_id]

        limit = pagin_config.limit if pagin_config else None
        if limit is None:
            limit = 10

        stream_token = yield self.store.get_stream_token_for_event(
            member_event_id
        )

        messages, token = yield self.store.get_recent_events_for_room(
            room_id,
            limit=limit,
            end_token=stream_token
        )

        messages = yield self._filter_events_for_client(
            user_id, messages, is_peeking=is_peeking
        )

        start_token = StreamToken.START.copy_and_replace("room_key", token[0])
        end_token = StreamToken.START.copy_and_replace("room_key", token[1])

        time_now = self.clock.time_msec()

        defer.returnValue({
            "membership": membership,
            "room_id": room_id,
            "messages": {
                "chunk": [serialize_event(m, time_now) for m in messages],
                "start": start_token.to_string(),
                "end": end_token.to_string(),
            },
            "state": [serialize_event(s, time_now) for s in room_state.values()],
            "presence": [],
            "receipts": [],
        })

    def on_GET(self, request, event_id):
        auth_user, _, _ = yield self.auth.get_user_by_req(request)
        handler = self.handlers.event_handler
        event = yield handler.get_event(auth_user, event_id)

        time_now = self.clock.time_msec()
        if event:
            defer.returnValue((200, serialize_event(event, time_now)))
        else:
            defer.returnValue((404, "Event not found."))

    def on_GET(self, request):
        requester = yield self.auth.get_user_by_req(request)
        user_id = requester.user.to_string()

        from_token = parse_string(request, "from", required=False)
        limit = parse_integer(request, "limit", default=50)
        only = parse_string(request, "only", required=False)

        limit = min(limit, 500)

        push_actions = yield self.store.get_push_actions_for_user(
            user_id, from_token, limit, only_highlight=(only == "highlight")
        )

        receipts_by_room = yield self.store.get_receipts_for_user_with_orderings(
            user_id, 'm.read'
        )

        notif_event_ids = [pa["event_id"] for pa in push_actions]
        notif_events = yield self.store.get_events(notif_event_ids)

        returned_push_actions = []

        next_token = None

        for pa in push_actions:
            returned_pa = {
                "room_id": pa["room_id"],
                "profile_tag": pa["profile_tag"],
                "actions": pa["actions"],
                "ts": pa["received_ts"],
                "event": serialize_event(
                    notif_events[pa["event_id"]],
                    self.clock.time_msec(),
                    event_format=format_event_for_client_v2_without_room_id,
                ),
            }

            if pa["room_id"] not in receipts_by_room:
                returned_pa["read"] = False
            else:
                receipt = receipts_by_room[pa["room_id"]]

                returned_pa["read"] = (
                    receipt["topological_ordering"], receipt["stream_ordering"]
                ) >= (
                    pa["topological_ordering"], pa["stream_ordering"]
                )
            returned_push_actions.append(returned_pa)
            next_token = str(pa["stream_ordering"])

        defer.returnValue((200, {
            "notifications": returned_push_actions,
            "next_token": next_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 on_GET(self, request, room_id):
        remote_server = urllib.unquote(
            request.args["remote"][0]
        ).decode("UTF-8")

        limit = int(request.args["limit"][0])

        handler = self.handlers.federation_handler
        events = yield handler.backfill(remote_server, room_id, limit)

        time_now = self.clock.time_msec()

        res = [serialize_event(event, time_now) for event in events]
        defer.returnValue((200, res))

    def on_GET(self, request, room_id, event_id):
        user, _, is_guest = yield self.auth.get_user_by_req(request, allow_guest=True)

        limit = int(request.args.get("limit", [10])[0])

        results = yield self.handlers.room_context_handler.get_event_context(
            user, room_id, event_id, limit, is_guest
        )

        time_now = self.clock.time_msec()
        results["events_before"] = [
            serialize_event(event, time_now) for event in results["events_before"]
        ]
        results["events_after"] = [
            serialize_event(event, time_now) for event in results["events_after"]
        ]
        results["state"] = [
            serialize_event(event, time_now) for event in results["state"]
        ]

        logger.info("Responding with %r", results)

        defer.returnValue((200, results))

    def get_stream(self, auth_user_id, pagin_config, timeout=0,
                   as_client_event=True, affect_presence=True,
                   only_room_events=False, room_id=None, is_guest=False):
        """Fetches the events stream for a given user.

        If `only_room_events` is `True` only room events will be returned.
        """
        auth_user = UserID.from_string(auth_user_id)

        try:
            if affect_presence:
                yield self.started_stream(auth_user)

            if timeout:
                # If they've set a timeout set a minimum limit.
                timeout = max(timeout, 500)

                # Add some randomness to this value to try and mitigate against
                # thundering herds on restart.
                timeout = random.randint(int(timeout*0.9), int(timeout*1.1))

            if is_guest:
                yield self.distributor.fire(
                    "user_joined_room", user=auth_user, room_id=room_id
                )

            events, tokens = yield self.notifier.get_events_for(
                auth_user, pagin_config, timeout,
                only_room_events=only_room_events,
                is_guest=is_guest, guest_room_id=room_id
            )

            time_now = self.clock.time_msec()

            chunks = [
                serialize_event(e, time_now, as_client_event) for e in events
            ]

            chunk = {
                "chunk": chunks,
                "start": tokens[0].to_string(),
                "end": tokens[1].to_string(),
            }

            defer.returnValue(chunk)

        finally:
            if affect_presence:
                self.stopped_stream(auth_user)

    def get_state_events(self, user_id, room_id):
        """Retrieve all state events for a given room.

        Args:
            user_id(str): The user requesting state events.
            room_id(str): The room ID to get all state events from.
        Returns:
            A list of dicts representing state events. [{}, {}, {}]
        """
        yield self.auth.check_joined_room(room_id, user_id)

        # TODO: This is duplicating logic from snapshot_all_rooms
        current_state = yield self.state_handler.get_current_state(room_id)
        now = self.clock.time_msec()
        defer.returnValue(
            [serialize_event(c, now) for c in current_state.values()]
        )

    def get_stream(self, auth_user_id, pagin_config, timeout=0,
                   as_client_event=True, affect_presence=True,
                   only_keys=None, room_id=None, is_guest=False):
        """Fetches the events stream for a given user.

        If `only_keys` is not None, events from keys will be sent down.
        """
        auth_user = UserID.from_string(auth_user_id)
        presence_handler = self.hs.get_presence_handler()

        context = yield presence_handler.user_syncing(
            auth_user_id, affect_presence=affect_presence,
        )
        with context:
            if timeout:
                # If they've set a timeout set a minimum limit.
                timeout = max(timeout, 500)

                # Add some randomness to this value to try and mitigate against
                # thundering herds on restart.
                timeout = random.randint(int(timeout * 0.9), int(timeout * 1.1))

            events, tokens = yield self.notifier.get_events_for(
                auth_user, pagin_config, timeout,
                only_keys=only_keys,
                is_guest=is_guest, explicit_room_id=room_id
            )

            # When the user joins a new room, or another user joins a currently
            # joined room, we need to send down presence for those users.
            to_add = []
            for event in events:
                if not isinstance(event, EventBase):
                    continue
                if event.type == EventTypes.Member:
                    if event.membership != Membership.JOIN:
                        continue
                    # Send down presence.
                    if event.state_key == auth_user_id:
                        # Send down presence for everyone in the room.
                        users = yield self.state.get_current_user_in_room(event.room_id)
                        states = yield presence_handler.get_states(
                            users,
                            as_event=True,
                        )
                        to_add.extend(states)
                    else:

                        ev = yield presence_handler.get_state(
                            UserID.from_string(event.state_key),
                            as_event=True,
                        )
                        to_add.append(ev)

            events.extend(to_add)

            time_now = self.clock.time_msec()

            chunks = [
                serialize_event(e, time_now, as_client_event) for e in events
            ]

            chunk = {
                "chunk": chunks,
                "start": tokens[0].to_string(),
                "end": tokens[1].to_string(),
            }

            defer.returnValue(chunk)

#.........这里部分代码省略.........

        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.
            state_filter = StateFilter.from_types(
                (EventTypes.Member, event.sender)
                for event in events
            )

            state_ids = yield self.store.get_state_ids_for_event(
                events[0].event_id, state_filter=state_filter,
            )

            if state_ids:
                state = yield self.store.get_events(list(state_ids.values()))
                state = state.values()

        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(),
        }

        if state:
            chunk["state"] = [
                serialize_event(e, time_now, as_client_event)
                for e in state
            ]

        defer.returnValue(chunk)

    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)