/usr/lib/python3/dist-packages/tftp/util.py is in python3-txtftp 0.1~bzr47-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | '''
@author: shylent
'''
from functools import wraps
from itertools import tee
from twisted.internet import reactor
from twisted.internet.defer import CancelledError, maybeDeferred
from twisted.internet.task import deferLater
__all__ = ['CANCELLED', 'deferred', 'timedCaller']
# Token used by L{timedCaller} to denote that it was cancelled instead of
# finishing naturally. This is not an error condition, and may not be of
# interest during normal use, but is helpful when testing.
CANCELLED = "CANCELLED"
def timedCaller(timings, call, last, clock=reactor):
"""Call C{call} or C{last} according to C{timings}.
The given C{timings} is an iterable of numbers. Each is a delay in seconds
that will be taken before making the next call to C{call} or C{last}.
The call to C{last} will happen after the last delay. If C{timings} is an
infinite iterable then C{last} will never be called.
This returns a C{Deferred} which can be cancelled. If the cancellation is
successful -- i.e. there is something to cancel -- then the result is set
to L{CANCELLED}. In other words, L{CancelledError} is squashed into a
non-failure condition.
@raise ValueError: if no timings are specified; there must be at least
one, even if specifies a zero seconds delay.
"""
timings = iterlast(timings)
def iterate(_=None):
for is_last, delay in timings:
# Return immediately with a deferred call.
if is_last:
return deferLater(clock, delay, last)
else:
return deferLater(clock, delay, call).addCallback(iterate)
else:
raise ValueError("No timings specified.")
def squashCancelled(failure):
if failure.check(CancelledError) is None:
return failure
else:
return CANCELLED
return iterate().addErrback(squashCancelled)
def iterlast(iterable):
"""Generate C{(is_last, item)} tuples from C{iterable}.
On each iteration this peeks ahead to see if the most recent iteration
will be the last, and returns this information as the C{is_last} element
of each tuple.
"""
iterable, peekable = tee(iterable)
next(peekable) # Advance in front.
for item in iterable:
try:
next(peekable)
except StopIteration:
yield True, item
else:
yield False, item
def deferred(func):
"""Decorates a function to ensure that it always returns a `Deferred`.
This also serves a secondary documentation purpose; functions decorated
with this are readily identifiable as asynchronous.
"""
@wraps(func)
def wrapper(*args, **kwargs):
return maybeDeferred(func, *args, **kwargs)
return wrapper
|