Python twisted.internet.reactor 模块,__class__() 实例源码

我们从Python开源项目中,提取了以下5个代码示例,用于说明如何使用twisted.internet.reactor.__class__()

项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def initialLog():
    from twisted.internet import reactor
    log.msg("twistd %s (%s %s) starting up" % (copyright.version,
                                               sys.executable,
                                               runtime.shortPythonVersion()))
    log.msg('reactor class: %s' % reactor.__class__)
项目:toughradius-benchmark    作者:toughmen    | 项目源码 | 文件源码
def install_reactor(explicitReactor=None, verbose=False):
    """
    Install Twisted reactor.

    :param explicitReactor: If provided, install this reactor. Else, install optimal reactor.
    :type explicitReactor: obj
    :param verbose: If ``True``, print what happens.
    :type verbose: bool
    """
    import sys

    if explicitReactor:
        ## install explicitly given reactor
        ##
        from twisted.application.reactors import installReactor
        print("Trying to install explicitly specified Twisted reactor '%s'" % explicitReactor)
        try:
            installReactor(explicitReactor)
        except Exception as e:
            print("Could not install Twisted reactor %s%s" % (explicitReactor, ' ["%s"]' % e if verbose else ''))
            sys.exit(1)
    else:
        ## automatically choose optimal reactor
        ##
        if verbose:
            print("Automatically choosing optimal Twisted reactor")
        install_optimal_reactor(verbose)

    ## now the reactor is installed, import it
    from twisted.internet import reactor

    if verbose:
        from twisted.python.reflect import qual
        print("Running Twisted reactor %s" % qual(reactor.__class__))

    return reactor
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def initialLog():
    from twisted.internet import reactor
    log.msg("twistd %s (%s %s) starting up" % (copyright.version,
                                               sys.executable,
                                               runtime.shortPythonVersion()))
    log.msg('reactor class: %s' % reactor.__class__)
项目:deb-python-autobahn    作者:openstack    | 项目源码 | 文件源码
def install_reactor(explicit_reactor=None, verbose=False):
    """
    Install Twisted reactor.

    :param explicit_reactor: If provided, install this reactor. Else, install
        the optimal reactor.
    :type explicit_reactor: obj
    :param verbose: If ``True``, print what happens.
    :type verbose: bool
    """
    import sys

    log = txaio.make_logger()

    if explicit_reactor:
        # install explicitly given reactor
        ##
        from twisted.application.reactors import installReactor
        log.info("Trying to install explicitly specified Twisted reactor '{reactor}'", reactor=explicit_reactor)
        try:
            installReactor(explicit_reactor)
        except:
            log.failure("Could not install Twisted reactor {reactor}\n{log_failure.value}",
                        reactor=explicit_reactor)
            sys.exit(1)
    else:
        # automatically choose optimal reactor
        ##
        log.debug("Automatically choosing optimal Twisted reactor")
        install_optimal_reactor(verbose)

    # now the reactor is installed, import it
    from twisted.internet import reactor
    txaio.config.loop = reactor

    if verbose:
        from twisted.python.reflect import qual
        log.debug("Running Twisted reactor {reactor}", reactor=qual(reactor.__class__))

    return reactor
项目:zenchmarks    作者:squeaky-pl    | 项目源码 | 文件源码
def _initialLog(self):
        """
        Print twistd start log message.
        """
        from twisted.internet import reactor
        logger._loggerFor(self).info(
            "twistd {version} ({exe} {pyVersion}) starting up.",
            version=copyright.version, exe=sys.executable,
            pyVersion=runtime.shortPythonVersion())
        logger._loggerFor(self).info('reactor class: {reactor}.',
                                     reactor=qual(reactor.__class__))