def cache(f):
    @wraps(f)
    @defer.inlineCallbacks
    def wrapper(*args, **kwargs):
        if 'cache_key' in kwargs and 'cache_ttl' in kwargs:
            key = "%s%s" % (f, kwargs['cache_key'])
            ttl = kwargs['cache_ttl']
            del kwargs['cache_key']
            del kwargs['cache_ttl']
            now = reactor.seconds()

            @defer.inlineCallbacks
            def get_value():
                result = yield f(*args, **kwargs)
                defer.returnValue(result)
            timestamp, result = CACHE.get(key, (0, None))
            if timestamp + ttl < now:
                CACHE[key] = (now, result)
                result = yield get_value()
                CACHE[key] = (now, result)
        else:
            result = yield f(*args, **kwargs)
        defer.returnValue(result)
    return wrapper 

def messageReceived(self, ignored, channel, message, nocache=False):
        try:
            json = anyjson.deserialize(message)
            db = self.factory.db
            db.last_data = reactor.seconds()
            pattern = json["pattern"]
            metric = json["metric"]
            yield db.addPatternMetric(pattern, metric)
            triggers = yield db.getPatternTriggers(pattern)
            if not triggers:
                yield db.removePattern(pattern)
                metrics = yield db.getPatternMetrics(pattern)
                for metric in metrics:
                    yield db.delMetric(metric)
                yield db.delPatternMetrics(pattern)

            for trigger_id in triggers:
                if nocache:
                    yield db.addTriggerCheck(trigger_id)
                else:
                    yield db.addTriggerCheck(trigger_id, cache_key=trigger_id, cache_ttl=config.CHECK_INTERVAL)
        except Exception as e:
            log.error("Failed to receive metric: {e}", e=e) 

def perform(self):
        try:
            trigger_id = yield self.db.getTriggerToCheck()
            while trigger_id is not None:
                acquired = yield self.db.setTriggerCheckLock(trigger_id)
                if acquired is not None:
                    start = reactor.seconds()
                    trigger = Trigger(trigger_id, self.db)
                    yield trigger.check()
                    end = reactor.seconds()
                    yield self.db.delTriggerCheckLock(trigger_id)
                    spy.TRIGGER_CHECK.report(end - start)
                trigger_id = yield self.db.getTriggerToCheck()
            yield task.deferLater(reactor, random.uniform(PERFORM_INTERVAL * 10, PERFORM_INTERVAL * 20), lambda: None)
        except GeneratorExit:
            pass
        except Exception as e:
            spy.TRIGGER_CHECK_ERRORS.report(0)
            log.error("Failed to perform triggers check: {e}", e=e)
            yield task.deferLater(reactor, ERROR_TIMEOUT, lambda: None) 

def test_startCheckingExpiration(self):
        """
        L{server.Session.startCheckingExpiration} causes the session to expire
        after L{server.Session.sessionTimeout} seconds without activity.
        """
        self.session.startCheckingExpiration()

        # Advance to almost the timeout - nothing should happen.
        self.clock.advance(self.session.sessionTimeout - 1)
        self.assertIn(self.uid, self.site.sessions)

        # Advance to the timeout, the session should expire.
        self.clock.advance(1)
        self.assertNotIn(self.uid, self.site.sessions)

        # There should be no calls left over, either.
        self.assertFalse(self.clock.calls) 

def test_nonASCII(self):
        """
        Bytes in fields of the request which are not part of ASCII are escaped
        in the result.
        """
        reactor = Clock()
        reactor.advance(1234567890)

        timestamp = http.datetimeToLogString(reactor.seconds())
        request = DummyRequestForLogTest(http.HTTPFactory(reactor=reactor))
        request.client = IPv4Address("TCP", b"evil x-forwarded-for \x80", 12345)
        request.method = b"POS\x81"
        request.protocol = b"HTTP/1.\x82"
        request.requestHeaders.addRawHeader(b"referer", b"evil \x83")
        request.requestHeaders.addRawHeader(b"user-agent", b"evil \x84")

        line = http.combinedLogFormatter(timestamp, request)
        self.assertEqual(
            u'"evil x-forwarded-for \\x80" - - [13/Feb/2009:23:31:30 +0000] '
            u'"POS\\x81 /dummy HTTP/1.0" 123 - "evil \\x83" "evil \\x84"',
            line) 

def _xforwardedforTest(self, header):
        """
        Assert that a request with the given value in its I{X-Forwarded-For}
        header is logged by L{proxiedLogFormatter} the same way it would have
        been logged by L{combinedLogFormatter} but with 172.16.1.2 as the
        client address instead of the normal value.

        @param header: An I{X-Forwarded-For} header with left-most address of
            172.16.1.2.
        """
        reactor = Clock()
        reactor.advance(1234567890)

        timestamp = http.datetimeToLogString(reactor.seconds())
        request = DummyRequestForLogTest(http.HTTPFactory(reactor=reactor))
        expected = http.combinedLogFormatter(timestamp, request).replace(
            u"1.2.3.4", u"172.16.1.2")
        request.requestHeaders.setRawHeaders(b"x-forwarded-for", [header])
        line = http.proxiedLogFormatter(timestamp, request)

        self.assertEqual(expected, line) 

def test_seconds(self):
        """
        L{twisted.internet.reactor.seconds} should return something
        like a number.

        1. This test specifically does not assert any relation to the
           "system time" as returned by L{time.time} or
           L{twisted.python.runtime.seconds}, because at some point we
           may find a better option for scheduling calls than
           wallclock-time.
        2. This test *also* does not assert anything about the type of
           the result, because operations may not return ints or
           floats: For example, datetime-datetime == timedelta(0).
        """
        now = reactor.seconds()
        self.assertEqual(now-now+now, now) 

def advanceCompletely(self, amount):
        """
        Move time on this clock forward by the given amount and run whatever
        pending calls should be run. Always complete the deferred calls before
        returning.

        @type amount: C{float}
        @param amount: The number of seconds which to advance this clock's
        time.
        """
        self.rightNow += amount
        self._sortCalls()
        while self.calls and self.calls[0].getTime() <= self.seconds():
            call = self.calls.pop(0)
            call.called = 1
            yield call.func(*call.args, **call.kw)
            self._sortCalls() 

def test_startCheckingExpiration(self):
        """
        L{server.Session.startCheckingExpiration} causes the session to expire
        after L{server.Session.sessionTimeout} seconds without activity.
        """
        self.session.startCheckingExpiration()

        # Advance to almost the timeout - nothing should happen.
        self.clock.advance(self.session.sessionTimeout - 1)
        self.assertIn(self.uid, self.site.sessions)

        # Advance to the timeout, the session should expire.
        self.clock.advance(1)
        self.assertNotIn(self.uid, self.site.sessions)

        # There should be no calls left over, either.
        self.assertFalse(self.clock.calls) 

def test_nonASCII(self):
        """
        Bytes in fields of the request which are not part of ASCII are escaped
        in the result.
        """
        reactor = Clock()
        reactor.advance(1234567890)

        timestamp = http.datetimeToLogString(reactor.seconds())
        request = DummyRequestForLogTest(http.HTTPFactory(reactor=reactor))
        request.client = IPv4Address("TCP", b"evil x-forwarded-for \x80", 12345)
        request.method = b"POS\x81"
        request.protocol = b"HTTP/1.\x82"
        request.requestHeaders.addRawHeader(b"referer", b"evil \x83")
        request.requestHeaders.addRawHeader(b"user-agent", b"evil \x84")

        line = http.combinedLogFormatter(timestamp, request)
        self.assertEqual(
            u'"evil x-forwarded-for \\x80" - - [13/Feb/2009:23:31:30 +0000] '
            u'"POS\\x81 /dummy HTTP/1.0" 123 - "evil \\x83" "evil \\x84"',
            line) 

def test_clientAddrIPv6(self):
        """
        A request from an IPv6 client is logged with that IP address.
        """
        reactor = Clock()
        reactor.advance(1234567890)

        timestamp = http.datetimeToLogString(reactor.seconds())
        request = DummyRequestForLogTest(http.HTTPFactory(reactor=reactor))
        request.client = IPv6Address("TCP", b"::1", 12345)

        line = http.combinedLogFormatter(timestamp, request)
        self.assertEqual(
            u'"::1" - - [13/Feb/2009:23:31:30 +0000] '
            u'"GET /dummy HTTP/1.0" 123 - "-" "-"',
            line) 

def test_clientAddrUnknown(self):
        """
        A request made from an unknown address type is logged as C{"-"}.
        """
        @implementer(interfaces.IAddress)
        class UnknowableAddress(object):
            """
            An L{IAddress} which L{combinedLogFormatter} cannot have
            foreknowledge of.
            """

        reactor = Clock()
        reactor.advance(1234567890)

        timestamp = http.datetimeToLogString(reactor.seconds())
        request = DummyRequestForLogTest(http.HTTPFactory(reactor=reactor))
        request.client = UnknowableAddress()

        line = http.combinedLogFormatter(timestamp, request)
        self.assertTrue(line.startswith(u'"-" ')) 

def _xforwardedforTest(self, header):
        """
        Assert that a request with the given value in its I{X-Forwarded-For}
        header is logged by L{proxiedLogFormatter} the same way it would have
        been logged by L{combinedLogFormatter} but with 172.16.1.2 as the
        client address instead of the normal value.

        @param header: An I{X-Forwarded-For} header with left-most address of
            172.16.1.2.
        """
        reactor = Clock()
        reactor.advance(1234567890)

        timestamp = http.datetimeToLogString(reactor.seconds())
        request = DummyRequestForLogTest(http.HTTPFactory(reactor=reactor))
        expected = http.combinedLogFormatter(timestamp, request).replace(
            u"1.2.3.4", u"172.16.1.2")
        request.requestHeaders.setRawHeaders(b"x-forwarded-for", [header])
        line = http.proxiedLogFormatter(timestamp, request)

        self.assertEqual(expected, line) 

def start(self):
        msg = bytearray('\0' * self.size)
        i = [0]

        def _sendmsg(c):
            if c > 1:
                log.msg('Packet source freeze for %d intervals at iter %d' % (c, i[0]))

            struct.pack_into('!HId', msg, 0, self.size, i[0], reactor.seconds())
            self.transport.write(msg, self.addr)
            i[0] += 1

            if i[0] >= self.count:
                lc.stop()

        lc = task.LoopingCall.withCount(_sendmsg)
        return lc.start(1.0 / self.rate, now=False) 

def test_seconds(self):
        """
        L{twisted.internet.reactor.seconds} should return something
        like a number.

        1. This test specifically does not assert any relation to the
           "system time" as returned by L{time.time} or
           L{twisted.python.runtime.seconds}, because at some point we
           may find a better option for scheduling calls than
           wallclock-time.
        2. This test *also* does not assert anything about the type of
           the result, because operations may not return ints or
           floats: For example, datetime-datetime == timedelta(0).
        """
        now = reactor.seconds()
        self.assertEquals(now-now+now, now) 

def add_ban(self, ip, reason, duration, name=None):
        """
        Ban an ip with an optional reason and duration in seconds. If duration
        is None, ban is permanent.
        """
        network = ip_network(str(ip), strict=False)
        for connection in list(self.connections.values()):
            if ip_address(connection.address[0]) in network:
                name = connection.name
                connection.kick(silent=True)
        if duration:
            duration = time.time() + duration
        else:
            duration = None
        self.bans[ip] = (name or '(unknown)', reason, duration)
        self.save_bans() 

def data_received(self, peer: Peer, packet: Packet) -> None:
        ip = peer.address.host
        current_time = reactor.seconds()
        try:
            ServerProtocol.data_received(self, peer, packet)
        except (NoDataLeft, ValueError):
            import traceback
            traceback.print_exc()
            log.info(
                'IP %s was hardbanned for invalid data or possibly DDoS.' % ip)
            self.hard_bans.add(ip)
            return
        dt = reactor.seconds() - current_time
        if dt > 1.0:
            log.warn('processing {!r} from {} took {}'.format(
                packet.data, ip, dt)) 

def update_world(self):
        last_time = self.last_time
        current_time = reactor.seconds()
        if last_time is not None:
            dt = current_time - last_time
            if dt > 1.0:
                log.warn('high CPU usage detected - %s' % dt)
        self.last_time = current_time
        ServerProtocol.update_world(self)
        time_taken = reactor.seconds() - current_time
        if time_taken > 1.0:
            log.warn(
                'World update iteration took %s, objects: %s' %
                (time_taken, self.world.objects))

    # events 

def start(cls, instigator, victim, reason=None):
        protocol = instigator.protocol
        last_votekick = instigator.last_votekick
        reason = reason.strip() if reason else None
        if protocol.votekick:
            raise VotekickFailure(S_IN_PROGRESS)
        elif instigator is victim:
            raise VotekickFailure(S_SELF_VOTEKICK)
        elif protocol.get_required_votes() <= 0:
            raise VotekickFailure(S_NOT_ENOUGH_PLAYERS)
        elif victim.admin or victim.rights.cancel or victim.local:
            raise VotekickFailure(S_VOTEKICK_IMMUNE)
        elif not instigator.admin and (last_votekick is not None and
                                       seconds() - last_votekick < cls.interval):
            raise VotekickFailure(S_NOT_YET)
        elif REQUIRE_REASON and not reason:
            raise VotekickFailure(S_NEED_REASON)

        result = protocol.on_votekick_start(instigator, victim, reason)
        if result is not None:
            raise VotekickFailure(result)

        reason = reason or S_DEFAULT_REASON
        return cls(instigator, victim, reason) 

def hit(self, value, by=None, kill_type=WEAPON_KILL):
        if self.hp is None:
            return
        if by is not None and self.team is by.team:
            friendly_fire = self.protocol.friendly_fire
            friendly_fire_on_grief = self.protocol.friendly_fire_on_grief
            if friendly_fire_on_grief:
                if (kill_type == MELEE_KILL and
                        not self.protocol.spade_teamkills_on_grief):
                    return
                hit_time = self.protocol.friendly_fire_time
                if (self.last_block_destroy is None
                        or reactor.seconds() - self.last_block_destroy >= hit_time):
                    return
            elif not friendly_fire:
                return
        self.set_hp(self.hp - value, by, kill_type=kill_type) 

def set_shoot(self, value: bool) -> None:
        if value == self.shoot:
            return
        current_time = reactor.seconds()
        if value:
            self.start = current_time
            if self.current_ammo <= 0:
                return
            elif self.reloading and not self.slow_reload:
                return
            self.shoot_time = max(current_time, self.next_shot)
            if self.reloading:
                self.reloading = False
                self.reload_call.cancel()
        else:
            ammo = self.current_ammo
            self.current_ammo = self.get_ammo(True)
            self.next_shot = self.shoot_time + self.delay * (
                ammo - self.current_ammo)
        self.shoot = value 

def call_belongs_to_internals(call):
    """Return True when `call` belongs to internal crochet and twisted..

    Crochet schedules a looping call that calls `reapAllProcesses` every 0.1
    seconds. This checks if this `DelayedCall` matches this signature.

    Twisted uses callLater in it's async reactor.

    :type call: :class:`DelayedCall`
    """
    if call.func.__module__ == "twisted.internet.asyncioreactor":
        return True
    elif isinstance(call.func, LoopingCall):
        return call.func.f is reapAllProcesses
    else:
        return False 

def _waitForThreadpoolToQuiesce(self, pool):
        """Return a :class:`Deferred` that waits for `pool` to quiesce."""
        now = reactor.seconds()
        until = now + 90.0
        while now < until:
            if self._isThreadpoolQuiet(pool):
                # The pool is quiet. It's safe to move on. Return the pool so
                # that it can still be reported.
                returnValue(pool)
            else:
                # Pause for a second to give it a chance to go quiet.
                now = yield deferLater(reactor, 1.0, reactor.seconds)
        else:
            # Despite waiting a long time the pool will not go quiet. The
            # validity of subsequent tests is compromised. Die immediately.
            print("ThreadPool", repr(pool), "is NOT quiet.", file=sys.stderr)
            os._exit(3) 

def get(self, timeout=None):
        """Get a promise for the value.

        Returns a `Deferred` that will fire with the value when it's made
        available, or with `CancelledError` if this object is cancelled.

        If a time-out in seconds is specified, the `Deferred` will be
        cancelled if the value is not made available within the time.
        """
        if self.waiters is None:
            return succeed(self.value)

        if timeout is None:
            d = Deferred()
        else:
            d = deferWithTimeout(timeout)

        def remove(result, discard, d):
            discard(d)  # Remove d from the waiters list.
            return result  # Pass-through the result.

        d.addBoth(remove, self.waiters.discard, d)
        self.waiters.add(d)
        return d 

def setUp(self):
        self.db = db.Db()
        self.db.rc = TwistedFakeRedis()
        yield self.db.startService()
        yield self.db.flush()
        datalib.db = self.db
        site = Site(self.db)
        self.protocol = MasterProtocol()
        self.protocol.factory = self
        self.port = reactor.listenTCP(0, site, interface="127.0.0.1")
        self.client = client.Agent(reactor)
        self.url_prefix = 'http://localhost:{0}{1}/'.format(
            self.port.getHost().port, site.prefix)
        self.now = int(reactor.seconds())
        self.check = TriggersCheck(self.db) 

def __init__(self, channel, queued):
        self.creation = reactor.seconds()
        self.body_json = None
        server.Request.__init__(self, channel, queued) 

def log(self, request):
        if hasattr(self, "logFile"):
            elapsed = reactor.seconds() - request.creation
            line = '- %.3f "%s" %d %s\n' % (
                elapsed,
                '%s %s %s' % (self._escape(request.method),
                              self._escape(request.uri),
                              self._escape(request.clientproto)),
                request.code,
                request.requestHeaders.getRawHeaders('Content-Length', ["-"])[0])
            self.logFile.write(line) 

def checkNoData(self):
        try:
            now = reactor.seconds()
            if self.db.last_data + config.STOP_CHECKING_INTERVAL < now:
                log.info("Checking nodata disabled. No metrics for {seconds} seconds",
                         seconds=int(now - self.db.last_data))
            else:
                log.info("Checking nodata")
                triggers = yield self.db.getTriggers()
                for trigger_id in triggers:
                    yield self.db.addTriggerCheck(trigger_id, cache_key=trigger_id, cache_ttl=60)
        except Exception as e:
            log.error("NoData check failed: {e}", e=e) 

def test_callLaterUsesReactorSecondsInDelayedCall(self):
        """
        L{reactor.callLater<twisted.internet.interfaces.IReactorTime.callLater>}
        should use the reactor's seconds factory
        to produce the time at which the DelayedCall will be called.
        """
        oseconds = reactor.seconds
        reactor.seconds = lambda: 100
        try:
            call = reactor.callLater(5, lambda: None)
            self.assertEqual(call.getTime(), 105)
        finally:
            reactor.seconds = oseconds
        call.cancel() 

def testDelayedCallSecondsOverride(self):
        """
        Test that the C{seconds} argument to DelayedCall gets used instead of
        the default timing function, if it is not None.
        """
        def seconds():
            return 10
        dc = base.DelayedCall(5, lambda: None, (), {}, lambda dc: None,
                              lambda dc: None, seconds)
        self.assertEqual(dc.getTime(), 5)
        dc.reset(3)
        self.assertEqual(dc.getTime(), 13)