test_threads.py 文件源码

def testCallFromThread(self):
        firedByReactorThread = defer.Deferred()
        firedByOtherThread = defer.Deferred()

        def threadedFunc():
            reactor.callFromThread(firedByOtherThread.callback, None)

        reactor.callInThread(threadedFunc)
        reactor.callFromThread(firedByReactorThread.callback, None)

        return defer.DeferredList(
            [firedByReactorThread, firedByOtherThread],
            fireOnOneErrback=True)