test_tcp.py 文件源码

def loopUntil(predicate, interval=0):
    """
    Poor excuse for an event notification helper.  This polls a condition and
    calls back a Deferred when it is seen to be true.

    Do not use this function.
    """
    from twisted.internet import task
    d = defer.Deferred()
    def check():
        res = predicate()
        if res:
            d.callback(res)
    call = task.LoopingCall(check)
    def stop(result):
        call.stop()
        return result
    d.addCallback(stop)
    d2 = call.start(interval)
    d2.addErrback(d.errback)
    return d