Python twisted.internet.reactor 模块，stop() 实例源码

def launch_crawlers(crawler_class, exclusion=None):
    settings = get_settings()
    configure_logging(settings=settings)
    launcher = CrawlerRunner(settings)
    crawlers = launcher.spider_loader.list()
    crawlers = list([c for c in crawlers if c.__contains__(crawler_class)])
    if exclusion:
        for c in settings.get(exclusion, []):
            crawlers.remove(c)

    try:
        for crawler in crawlers:
            launcher.crawl(crawler)
        d = launcher.join()
        d.addBoth(lambda _: reactor.stop())
        reactor.run()
        return True
    except Exception as e:
        launch_logger.error('(????)????? | ?????:\n{excep}'
                            .format(excep=e))
        return False

def fixPdb():
    def do_stop(self, arg):
        self.clear_all_breaks()
        self.set_continue()
        from twisted.internet import reactor
        reactor.callLater(0, reactor.stop)
        return 1

    def help_stop(self):
        print """stop - Continue execution, then cleanly shutdown the twisted reactor."""

    def set_quit(self):
        os._exit(0)

    pdb.Pdb.set_quit = set_quit
    pdb.Pdb.do_stop = do_stop
    pdb.Pdb.help_stop = help_stop

def run():
#    import hotshot
#    prof = hotshot.Profile('cftp.prof')
#    prof.start()
    args = sys.argv[1:]
    if '-l' in args: # cvs is an idiot
        i = args.index('-l')
        args = args[i:i+2]+args
        del args[i+2:i+4]
    options = ClientOptions()
    try:
        options.parseOptions(args)
    except usage.UsageError, u:
        print 'ERROR: %s' % u
        sys.exit(1)
    if options['log']:
        realout = sys.stdout
        log.startLogging(sys.stderr)
        sys.stdout = realout
    else:
        log.discardLogs()
    doConnect(options)
    reactor.run()
#    prof.stop()
#    prof.close()

def getPrivateKey(self):
        file = os.path.expanduser(self.usedFiles[-1])
        if not os.path.exists(file):
            return None
        try:
            return defer.succeed(keys.getPrivateKeyObject(file))
        except keys.BadKeyError, e:
            if e.args[0] == 'encrypted key with no passphrase':
                for i in range(3):
                    prompt = "Enter passphrase for key '%s': " % \
                           self.usedFiles[-1]
                    try:
                        p = self._getPassword(prompt)
                        return defer.succeed(keys.getPrivateKeyObject(file, passphrase = p))
                    except (keys.BadKeyError, ConchError):
                        pass
                return defer.fail(ConchError('bad password'))
            raise
        except KeyboardInterrupt:
            print
            reactor.stop()

def test_crash(self):
        """
        reactor.crash should NOT fire shutdown triggers
        """
        events = []
        self.addTrigger(
            "before", "shutdown",
            lambda: events.append(("before", "shutdown")))

        # reactor.crash called from an "after-startup" trigger is too early
        # for the gtkreactor: gtk_mainloop is not yet running. Same is true
        # when called with reactor.callLater(0). Must be >0 seconds in the
        # future to let gtk_mainloop start first.
        reactor.callWhenRunning(
            reactor.callLater, 0, reactor.crash)
        reactor.run()
        self.failIf(events, "reactor.crash invoked shutdown triggers, but it "
                            "isn't supposed to.")

    # XXX Test that reactor.stop() invokes shutdown triggers

def stop_reactor():
    """Stop the reactor and join the reactor thread until it stops.
    Call this function in teardown at the module or package level to
    reset the twisted system after your tests. You *must* do this if
    you mix tests using these tools and tests using twisted.trial.
    """
    global _twisted_thread

    def stop_reactor():
        '''Helper for calling stop from withing the thread.'''
        reactor.stop()

    reactor.callFromThread(stop_reactor)
    reactor_thread.join()
    for p in reactor.getDelayedCalls():
        if p.active():
            p.cancel()
    _twisted_thread = None

def startService(self):
        """Construct server and bind."""
        from twisted.internet import reactor
        service.Service.startService(self)

        def connected(port):
            self._port = port

        # noinspection PyUnresolvedReferences
        def failure(err):
            log.err(err, _why='Could not bind to port')
            reactor.stop()

        factory = self._factory_klass()
        server = endpoints.serverFromString(reactor, self._endpoint)

        d = server.listen(factory)
        d.addCallbacks(connected, failure)
        return d

def crawl(args):
    spids = args.get('spiders')
    configure_logging(SETTINGS,
                      install_root_handler=False)
    logging.getLogger('scrapy').setLevel(logging.WARNING)
    runner = CrawlerRunner(SETTINGS)
    loader = runner.spider_loader
    if 'all' in spids:
        spids = loader.list()
    spiders = [loader.load(_)
               for _ in filter(lambda __: __ in loader.list(),
                               spids)]
    if not spiders:
        return False

    random.shuffle(spiders)
    for __ in spiders:
        runner.crawl(__)
    d = runner.join()
    d.addBoth(lambda _: reactor.stop())

    logger.info('crawl reator starting ...')
    reactor.run()
    logging.info('crawl reator stopped')

def test_ping_failure_counter_reset_after_restart(self):
        """
        When a daemon stops responding and gets restarted after 5 failed pings,
        it will wait for another 5 failed pings before it will be restarted
        again.
        """
        clock = Clock()
        dog = WatchDog(clock,
                       broker=AsynchronousPingDaemon("test-broker"),
                       monitor=BoringDaemon("test-monitor"),
                       manager=BoringDaemon("test-manager"))
        dog.start_monitoring()

        for i in range(5):
            clock.advance(5)
            dog.broker.fire_running(False)

        self.assertEqual(dog.broker.boots, ["stop", "start"])
        for i in range(4):
            clock.advance(5)
            dog.broker.fire_running(False)
            self.assertEqual(dog.broker.boots, ["stop", "start"])
        clock.advance(5)
        dog.broker.fire_running(False)
        self.assertEqual(dog.broker.boots, ["stop", "start", "stop", "start"])

def disconnected(self, daemon):
        self.mainwin.set_sensitive(False)
        # If the reactor is not running at this point it means that we were
        # closed normally.
        if not reactor.running:
            return
        self.save_settings()
        msg = _("Lost connection with the epoptes service.")
        msg += "\n\n" + _("Make sure the service is running and then restart epoptes.")
        dlg = gtk.MessageDialog(type=gtk.MESSAGE_ERROR, buttons=gtk.BUTTONS_OK,
                                 message_format=msg)
        dlg.set_title(_('Service connection error'))
        dlg.run()
        dlg.destroy()
        reactor.stop()


    # AMP callbacks

def poetry_main():
    addresses = parser_args()  # ??????IP???,???????,
    from twisted.internet import reactor
    poems = []  # ???????,?????????

    def got_poem(poem):  # ????,?????????????????
        poems.append(poem)  # ??????????
        if len(poems) == len(addresses):  # ??????????????????????????,????????,
            reactor.stop()  # ??reactor?stop?????????????

    for address in addresses:
        host, port = address  # ?????????????,??ip,port
        get_poetry(host=host, port=port, callback=got_poem)  # ????????????ip,port??,?????got_poem????
    reactor.run()  # ??????,???????,??????select???...

    print("main loop done...")  # ???????????,?????????

def __init__(self,server,port,secret,requests,concurrency,username,password,
        verb=False,timeout=30,forknum=1,interval=2,rate=1000):
        self.interval = interval
        tparams = [
            ' - Client platform                   :  %s, %s'% (platform.platform(),platform.machine()),
            ' - Python implement, version         :  %s, %s'% (platform.python_implementation(), platform.python_version()),
            ' - Radius server  address            :  %s'% server,
            ' - Radius Server auth port           :  %s'% port,
            ' - Raduius share secret              :  %s'% secret,
            ' - Auth Request total                :  %s'% requests,
            ' - Concurrency level                 :  %s'% concurrency,
            ' - Worker Process num                :  %s'% forknum,
            ' - All Requests timeout              :  %s sec'% timeout,
            ' - Stat data interval                :  %s sec'% interval,
            ' - Send request rate                 :  %s/sec'% rate,
        ]
        self.stat_counter = AuthStatCounter(tparams)
        self.puller = ZmqPullConnection(ZmqFactory(), ZmqEndpoint('bind', 'ipc:///tmp/toughbt-message'))
        self.puller.onPull = self.do_stat
        # log.msg("init BenchmarkMaster puller : %s " % (self.puller))

        reactor.callLater(interval,self.chk_task)
        reactor.callLater(timeout,reactor.stop)

def fixPdb():
    def do_stop(self, arg):
        self.clear_all_breaks()
        self.set_continue()
        from twisted.internet import reactor
        reactor.callLater(0, reactor.stop)
        return 1

    def help_stop(self):
        print """stop - Continue execution, then cleanly shutdown the twisted reactor."""

    def set_quit(self):
        os._exit(0)

    pdb.Pdb.set_quit = set_quit
    pdb.Pdb.do_stop = do_stop
    pdb.Pdb.help_stop = help_stop

def run():
#    import hotshot
#    prof = hotshot.Profile('cftp.prof')
#    prof.start()
    args = sys.argv[1:]
    if '-l' in args: # cvs is an idiot
        i = args.index('-l')
        args = args[i:i+2]+args
        del args[i+2:i+4]
    options = ClientOptions()
    try:
        options.parseOptions(args)
    except usage.UsageError, u:
        print 'ERROR: %s' % u
        sys.exit(1)
    if options['log']:
        realout = sys.stdout
        log.startLogging(sys.stderr)
        sys.stdout = realout
    else:
        log.discardLogs()
    doConnect(options)
    reactor.run()
#    prof.stop()
#    prof.close()

def getPrivateKey(self):
        file = os.path.expanduser(self.usedFiles[-1])
        if not os.path.exists(file):
            return None
        try:
            return defer.succeed(keys.getPrivateKeyObject(file))
        except keys.BadKeyError, e:
            if e.args[0] == 'encrypted key with no passphrase':
                for i in range(3):
                    prompt = "Enter passphrase for key '%s': " % \
                           self.usedFiles[-1]
                    try:
                        p = self._getPassword(prompt)
                        return defer.succeed(keys.getPrivateKeyObject(file, passphrase = p))
                    except (keys.BadKeyError, ConchError):
                        pass
                return defer.fail(ConchError('bad password'))
            raise
        except KeyboardInterrupt:
            print
            reactor.stop()

def test_crash(self):
        """
        reactor.crash should NOT fire shutdown triggers
        """
        events = []
        self.addTrigger(
            "before", "shutdown",
            lambda: events.append(("before", "shutdown")))

        # reactor.crash called from an "after-startup" trigger is too early
        # for the gtkreactor: gtk_mainloop is not yet running. Same is true
        # when called with reactor.callLater(0). Must be >0 seconds in the
        # future to let gtk_mainloop start first.
        reactor.callWhenRunning(
            reactor.callLater, 0, reactor.crash)
        reactor.run()
        self.failIf(events, "reactor.crash invoked shutdown triggers, but it "
                            "isn't supposed to.")

    # XXX Test that reactor.stop() invokes shutdown triggers

def stop_reactor():
    """Stop the reactor and join the reactor thread until it stops.
    Call this function in teardown at the module or package level to
    reset the twisted system after your tests. You *must* do this if
    you mix tests using these tools and tests using twisted.trial.
    """
    global _twisted_thread

    def stop_reactor():
        '''Helper for calling stop from withing the thread.'''
        reactor.stop()

    reactor.callFromThread(stop_reactor)
    reactor_thread.join()
    for p in reactor.getDelayedCalls():
        if p.active():
            p.cancel()
    _twisted_thread = None

def check_for_phase1_utxos(self, utxos, cb=None):
        """Any participant needs to wait for completion of phase 1 through
        seeing the utxos on the network. Optionally pass callback for start
        of phase2 (redemption phase), else default is state machine tick();
        must have signature callback(utxolist).
        Triggered on number of confirmations as set by config.
        This should be fired by task looptask, which is stopped on success.
        """
        result = cs_single().bc_interface.query_utxo_set(utxos,
                                                         includeconf=True)
        if None in result:
            return
        for u in result:
            if u['confirms'] < self.coinswap_parameters.tx01_confirm_wait:
                return
        self.loop.stop()
        if cb:
            cb()
        else:
            self.sm.tick()

def stop_reaktor(_):
    reactor.stop()

def stop(self):
        """Call to cleanup the reactor"""
        ThreadCommand((self._doStop, (), {}), self._onStopped, self._onStopFailed)

def _onStopFailed(self, reason):
        self.running = False
        raise Exception('Could not stop reactor: %s' % reason)

def _doStop(self, tc):
        reactor.stop()

def stop(self, onStopped, onFail):
        ThreadCommand((self._doStop, (), {}), onStopped, onFail)
        self.started = False

def onClose(self, evt):
        """Stops the twisted threads"""
        self.twistedThread.stop()
        evt.Skip()

def clientConnectionLost(self, connector, reason):
         reactor.stop()

def stop_reactor(exit_code=0):
    '''
    Stop the reactor and exit with exit_code.
    If exit_code is None, don't exit, just return to the caller.
    exit_code must be between 1 and 255.
    '''
    if exit_code is not None:
        logging.warning("Exiting with code {}".format(exit_code))
    else:
        # Let's hope the calling code exits pretty soon after this
        logging.warning("Stopping reactor")

    try:
        reactor.stop()
    except ReactorNotRunning:
        pass

    # return to the caller and let it decide what to do
    if exit_code == None:
        return

    # a graceful exit
    if exit_code == 0:
        sys.exit()

    # a hard exit
    assert exit_code >= 0
    assert exit_code <= 127
    os._exit(exit_code)

def clientConnectionLost(self, connector, reason):
        print 'Connection lost. Reason:', reason
        if reactor.running:
            reactor.stop()

def stop(self):
        LOG.info("Stopping AMQP server... ")
        self.server.stop()
        LOG.info("AMQP server stopped. ")

def stop(self):
        LOG.info("Stopping WAMP-agent server...")
        reactor.stop()
        LOG.info("WAMP server stopped.")

def stop_handler(self, signum, frame):
        self.w.stop()
        self.r.stop()
        self.del_host()
        os._exit(0)

def _stop_reactor(self, _=None):
        reactor.stop()

def doItTkinterly(opt):
    root=Tkinter.Tk()
    root.withdraw()
    root.title('One Moment.')
    root.protocol('WM_DELETE_WINDOW', reactor.stop)
    tksupport.install(root)

    prog=ProgressBar(root, value=0, labelColor="black", width=200)
    prog.pack()

    # callback immediately
    d=defer.succeed(root).addErrback(log.err)

    def deiconify(root):
        root.deiconify()
        return root

    d.addCallback(deiconify)

    if opt['zipfile']:
        uz=Progressor('Unpacking documentation...')
        max=zipstream.countZipFileChunks(opt['zipfile'], 4096)
        uz.setBar(prog, max)
        uz.setIterator(zipstream.unzipIterChunky(opt['zipfile'],
                                                 opt['ziptargetdir']))
        d.addCallback(uz.processAll)

    if opt['compiledir']:
        comp=Progressor('Compiling to pyc...')
        comp.setBar(prog, countPysRecursive(opt['compiledir']))
        comp.setIterator(compiler(opt['compiledir']))
        d.addCallback(comp.processAll)

    def stop(ignore):
        reactor.stop()
        root.destroy()
    d.addCallback(stop)

    reactor.run()

def handle_QUIT(self):
        ColoredManhole.handle_QUIT(self)
        reactor.stop()

def deferredAskFrame(question, echo):
    if frame.callback:
        raise "can't ask 2 questions at once!"
    d = defer.Deferred()
    resp = []
    def gotChar(ch, resp=resp):
        if not ch: return
        if ch=='\x03': # C-c
            reactor.stop()
        if ch=='\r':
            frame.write('\r\n')
            stresp = ''.join(resp)
            del resp
            frame.callback = None
            d.callback(stresp)
            return
        elif 32 <= ord(ch) < 127:
            resp.append(ch)
            if echo:
                frame.write(ch)
        elif ord(ch) == 8 and resp: # BS
            if echo: frame.write('\x08 \x08')
            resp.pop()
    frame.callback = gotChar
    frame.write(question)
    frame.canvas.focus_force()
    return d

def handleError():
    from twisted.python import failure
    global exitStatus
    exitStatus = 2
    log.err(failure.Failure())
    reactor.stop()
    raise

def closed(self):
        log.msg('closed %s' % self)
        if len(self.conn.channels) == 1: # just us left
            reactor.stop()

def handleError():
    from twisted.python import failure
    global exitStatus
    exitStatus = 2
    try:
        reactor.stop()
    except: pass
    log.err(failure.Failure())
    raise

def _ebExit(f):
    #global exitStatus
    if hasattr(f.value, 'value'):
        s = f.value.value
    else:
        s = str(f)
    print s
    #exitStatus = "conch: exiting with error %s" % f
    try:
        reactor.stop()
    except: pass

def _stopReactor():
    try:
        reactor.stop()
    except: pass

def _guiUpdate(reactor, delay):
    pyui.draw()
    if pyui.update() == 0:
        pyui.quit()
        reactor.stop()
    else:
        reactor.callLater(delay, _guiUpdate, reactor, delay)

def connectionLost(self, reason):
        reactor.stop()

def connectionLost(self, reason):
        reactor.stop()

def testIterate(self):
        """
        Test that reactor.iterate(0) doesn't block
        """
        start = time.time()
        # twisted timers are distinct from the underlying event loop's
        # timers, so this fail-safe probably won't keep a failure from
        # hanging the test
        t = reactor.callLater(10, reactor.crash)
        reactor.iterate(0) # shouldn't block
        stop = time.time()
        elapsed = stop - start
        #print "elapsed", elapsed
        self.failUnless(elapsed < 8)
        t.cancel()

def stopProducing(self):
        self.events.append('stop')

def test_unconnectedFileDescriptor(self):
        """
        Verify that registering a producer when the connection has already
        been closed invokes its stopProducing() method.
        """
        fd = abstract.FileDescriptor()
        fd.disconnected = 1
        dp = DummyProducer()
        fd.registerProducer(dp, 0)
        self.assertEquals(dp.events, ['stop'])

def testShutdownFromDatagramReceived(self):
        """Test reactor shutdown while in a recvfrom() loop"""

        # udp.Port's doRead calls recvfrom() in a loop, as an optimization.
        # It is important this loop terminate under various conditions.
        # Previously, if datagramReceived synchronously invoked
        # reactor.stop(), under certain reactors, the Port's socket would
        # synchronously disappear, causing an AttributeError inside that
        # loop.  This was mishandled, causing the loop to spin forever.
        # This test is primarily to ensure that the loop never spins
        # forever.

        finished = defer.Deferred()
        pr = self.server.packetReceived = defer.Deferred()

        def pktRece(ignored):
            # Simulate reactor.stop() behavior :(
            self.server.transport.connectionLost()
            # Then delay this Deferred chain until the protocol has been
            # disconnected, as the reactor should do in an error condition
            # such as we are inducing.  This is very much a whitebox test.
            reactor.callLater(0, finished.callback, None)
        pr.addCallback(pktRece)

        def flushErrors(ignored):
            # We are breaking abstraction and calling private APIs, any
            # number of horrible errors might occur.  As long as the reactor
            # doesn't hang, this test is satisfied.  (There may be room for
            # another, stricter test.)
            self.flushLoggedErrors()
        finished.addCallback(flushErrors)
        self.server.transport.write('\0' * 64, ('127.0.0.1',
                                    self.server.transport.getHost().port))
        return finished