/usr/lib/python2.7/dist-packages/foolscap/util.py is in python-foolscap 0.6.4-2.
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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | import socket
import time
from twisted.internet import defer, reactor, protocol
class AsyncAND(defer.Deferred):
"""Like DeferredList, but results are discarded and failures handled
in a more convenient fashion.
Create me with a list of Deferreds. I will fire my callback (with None)
if and when all of my component Deferreds fire successfully. I will fire
my errback when and if any of my component Deferreds errbacks, in which
case I will absorb the failure. If a second Deferred errbacks, I will not
absorb that failure.
This means that you can put a bunch of Deferreds together into an
AsyncAND and then forget about them. If all succeed, the AsyncAND will
fire. If one fails, that Failure will be propagated to the AsyncAND. If
multiple ones fail, the first Failure will go to the AsyncAND and the
rest will be left unhandled (and therefore logged).
"""
def __init__(self, deferredList):
defer.Deferred.__init__(self)
if not deferredList:
self.callback(None)
return
self.remaining = len(deferredList)
self._fired = False
for d in deferredList:
d.addCallbacks(self._cbDeferred, self._cbDeferred,
callbackArgs=(True,), errbackArgs=(False,))
def _cbDeferred(self, result, succeeded):
self.remaining -= 1
if succeeded:
if not self._fired and self.remaining == 0:
# the last input has fired. We fire.
self._fired = True
self.callback(None)
return
else:
if not self._fired:
# the first Failure is carried into our output
self._fired = True
self.errback(result)
return None
else:
# second and later Failures are not absorbed
return result
# adapted from Tahoe: finds a single publically-visible address, or None.
# Tahoe also uses code to run /bin/ifconfig (or equivalent) to find other
# addresses, but that's a bit heavy for this. Note that this runs
# synchronously. Also note that this doesn't require the reactor to be
# running.
def get_local_ip_for(target='A.ROOT-SERVERS.NET'):
"""Find out what our IP address is for use by a given target.
@return: the IP address as a dotted-quad string which could be used by
to connect to us. It might work for them, it might not. If
there is no suitable address (perhaps we don't currently have an
externally-visible interface), this will return None.
"""
try:
target_ipaddr = socket.gethostbyname(target)
except socket.gaierror:
# DNS isn't running
return None
udpprot = protocol.DatagramProtocol()
port = reactor.listenUDP(0, udpprot)
try:
udpprot.transport.connect(target_ipaddr, 7)
localip = udpprot.transport.getHost().host
except socket.error:
# no route to that host
localip = None
port.stopListening() # note, this returns a Deferred
return localip
FORMAT_TIME_MODES = ["short-local", "long-local", "utc", "epoch"]
def format_time(when, mode):
if mode == "short-local":
time_s = time.strftime("%H:%M:%S", time.localtime(when))
time_s = time_s + ".%03d" % int(1000*(when - int(when)))
elif mode == "long-local":
lt = time.localtime(when)
time_s = time.strftime("%Y-%m-%d_%H:%M:%S", lt)
time_s = time_s + ".%06d" % int(1000000*(when - int(when)))
time_s += time.strftime("%z", lt)
elif mode == "utc":
time_s = time.strftime("%Y-%m-%d_%H:%M:%S", time.gmtime(when))
time_s = time_s + ".%06d" % int(1000000*(when - int(when)))
time_s += "Z"
elif mode == "epoch":
time_s = "%.03f" % when
return time_s
|