<fim-prefix>scii")
                # If the header exists, add to the comma-separated list of the first
                # instance of the header. Otherwise, generate a new header.
                if x_forwarded_for:
                    x_forwarded_for = [
                        x_forwarded_for[0] + b", " + previous_host
                    ] + x_forwarded_for[1:]
                else:
                    x_forwarded_for = [previous_host]
            headers[b"X-Forwarded-For"] = x_forwarded_for

            try:
                result = await self.http_client.post_json_get_json(
                    self.main_uri + request.uri.decode("ascii"), body, headers=headers
                )
            except HttpResponseException as e:
                raise e.to_synapse_error() from e
            except RequestSendFailed as e:
                raise SynapseError(502, "Failed to talk to master") from e

            return 200, result
        else:
            # Just interested in counts.
            result = await self.store.count_e2e_one_time_keys(user_id, device_id)
            return 200, {"one_time_key_counts": result}


class _NullContextManager(ContextManager[None]):
    """A context manager which does nothing."""

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


UPDATE_SYNCING_USERS_MS = 10 * 1000


class GenericWorkerPresence(BasePresenceHandler):
    def __init__(self, hs):
        super().__init__(hs)
        self.hs = hs
        self.is_mine_id = hs.is_mine_id
        # BUG: CWE-601 URL Redirection to Untrusted Site ('Open Redirect')
        #         self.http_client = hs.get_simple_http_client()
        # FIXED: 
 <fim-middle>       self._presence_enabled = hs.config.use_presence

        # The number of ongoing syncs on this process, by user id.
        # Empty if _presence_enabled is false.
        self._user_to_num_curre<fix-suffix>nt_syncs = {}  # type: Dict[str, int]

        self.notifier = hs.get_notifier()
        self.instance_id = hs.get_instance_id()

        # user_id -> last_sync_ms. Lists the users that have stopped syncing
        # but we haven't notified the master of that yet
        self.users_going_offline = {}

        self._bump_active_client = ReplicationBumpPresenceActiveTime.make_client(hs)
        self._set_state_client = ReplicationPresenceSetState.make_client(hs)

        self._send_stop_syncing_loop = self.clock.looping_call(
            self.send_stop_syncing, UPDATE_SYNCING_USERS_MS
        )

        hs.get_reactor().addSystemEventTrigger(
            "before",
            "shutdown",
            run_as_background_process,
            "generic_presence.on_shutdown",
            self._on_shutdown,
        )

    def _on_shutdown(self):
        if self._presence_enabled:
            self.hs.get_tcp_replication().send_command(
                ClearUserSyncsCommand(self.instance_id)
     <|endoftext|>