Python asyncio 模块，ProactorEventLoop() 实例源码

def __init__(self, loop=None):
        """Initialize new Home Assistant object."""
        if sys.platform == "win32":
            self.loop = loop or asyncio.ProactorEventLoop()
        else:
            self.loop = loop or asyncio.get_event_loop()

        self.executor = ThreadPoolExecutor(max_workers=EXECUTOR_POOL_SIZE)
        self.loop.set_default_executor(self.executor)
        self.loop.set_exception_handler(self._async_exception_handler)
        self._pending_tasks = []
        self.bus = EventBus(self)
        self.services = ServiceRegistry(self)
        self.states = StateMachine(self.bus, self.loop)
        self.config = Config()  # type: Config
        # This is a dictionary that any component can store any data on.
        self.data = {}
        self.state = CoreState.not_running
        self.exit_code = None

def main():
    if os.name == 'nt':
        loop = asyncio.ProactorEventLoop()
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()
    loop.run_until_complete(start(
        'sleep 2; wc', input=[b'foo bar baz\n'*300 for i in range(100)]))
    loop.close()

def create_event_loop(self):
            return asyncio.ProactorEventLoop()

def setUp(self):
            self.loop = asyncio.ProactorEventLoop()
            self.set_event_loop(self.loop)

def setUp(self):
        self.loop = asyncio.ProactorEventLoop()
        self.set_event_loop(self.loop)

def aiohttp_tarantool_multi(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start aiohttp_tarantool_multi_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start aiohttp_tarantool_multi")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from data_interface.tarantool_driver import TarantoolDriver
    from aiohttp_server.aiohttpmulticonnect import AiohttpTarantoolMulti
    db = TarantoolDriver
    web_server = AiohttpTarantoolMulti(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def aiohttp_tarantool_one(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start aiohttp_tarantool_one_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start aiohttp_tarantool_one")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from data_interface.tarantool_driver import TarantoolDriver
    from aiohttp_server.aiohttponeconnect import AiohttpUniversalOne
    db = TarantoolDriver()
    web_server = AiohttpUniversalOne(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def aiohttp_postgres_pool(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start aiohttp_postgres_pool_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start aiohttp_postgres_pool")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)
    from data_interface.postgres_driver import PostgresDriver
    from aiohttp_server.aiohttponeconnect import AiohttpUniversalOne
    pd = PostgresDriver(loop)
    db = loop.run_until_complete(pd.open())
    web_server = AiohttpUniversalOne(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def sanic_tarantool_one(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start sanic_tarantool_one_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start sanic_tarantool_one")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from sanic_server.saniconeconnect import SanicTarantoolOne
    from data_interface.tarantool_driver import TarantoolDriver
    db = TarantoolDriver()
    web_server = SanicTarantoolOne(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def japronto_tarantool_one(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start japronto_tarantool_one_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start japronto_tarantool_one")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from japronto_server import janprontooneconnect
    from data_interface.tarantool_driver import TarantoolDriver
    db = TarantoolDriver()
    janprontooneconnect.init(loop_param=loop, port_param=port, db_driver=db)
    loop.create_task(janprontooneconnect.listner_for_statistics())
    janprontooneconnect.start()
    # Return false for compatibility japronto implementation with other
    # servers
    # Japronto hard coded to create own loop and run_forever it :(
    return False  # It will raise Exeptions

def broker(ctx, in_connect, in_bind, out_connect, out_bind):
    loop = LoopClass()
    context = azmq.Context(loop=loop)
    in_socket = context.socket(azmq.PULL)
    out_socket = context.socket(azmq.PUSH)

    if in_connect:
        click.echo("Incoming connecting to %s." % in_connect, err=True)
        in_socket.connect(in_connect)
    else:
        click.echo("Incoming binding on %s." % in_bind, err=True)
        in_socket.bind(in_bind)

    if out_connect:
        click.echo("Outgoing connecting to %s." % out_connect, err=True)
        out_socket.connect(out_connect)
    else:
        click.echo("Outgoing binding on %s." % out_bind, err=True)
        out_socket.bind(out_bind)

    async def run(context, in_socket, out_socket):
        click.echo("Broker started.", err=True)

        try:
            while True:
                msg = await in_socket.recv_multipart()
                await out_socket.send_multipart(msg)
        finally:
            click.echo("Broker stopped.", err=True)

    task = asyncio.ensure_future(
        run(context, in_socket, out_socket),
        loop=loop,
    )

    with allow_interruption((loop, context.close)):
        loop.run_until_complete(context.wait_closed())

    loop.close()

def event_loop():
    if sys.platform == 'win32':
        loop = asyncio.ProactorEventLoop()
    else:
        loop = asyncio.SelectorEventLoop()

    asyncio.set_event_loop(loop)

    yield loop

    tasks = [task for task in asyncio.Task.all_tasks() if not task.done()]

    if tasks:
        loop.run_until_complete(asyncio.wait_for(asyncio.wait(tasks), 5))

def check_event_loop():
    if sys.platform == 'win32' and isinstance(asyncio.get_event_loop(), asyncio.SelectorEventLoop):
        raise ValueError(
            'SelectorEventLoop is not supported on Windows, use asyncio.ProactorEventLoop instead.'
        )

def create_event_loop(self):
            return asyncio.ProactorEventLoop()

def setUp(self):
            self.loop = asyncio.ProactorEventLoop()
            self.set_event_loop(self.loop)

def setUp(self):
        self.loop = asyncio.ProactorEventLoop()
        self.set_event_loop(self.loop)

def create_event_loop(self):
            return asyncio.ProactorEventLoop()

def setUp(self):
            policy = asyncio.get_event_loop_policy()
            self.loop = asyncio.ProactorEventLoop()

            # ensure that the event loop is passed explicitly in asyncio
            policy.set_event_loop(None)

def setUp(self):
        self.loop = asyncio.ProactorEventLoop()
        self.set_event_loop(self.loop)

def main():  # pragma: no cover
    if sys.platform == "win32":
        _loop = asyncio.ProactorEventLoop()
        asyncio.set_event_loop(_loop)
    else:
        _loop = asyncio.get_event_loop()
    try:
        main_loop(_loop)
    finally:
        _loop.close()

def _start_controller(self, loop):
        """
        Control structure to start new cmd
        :param loop: event loop
        :return: None
        """
        # Loop check
        if loop is None:
            if sys.platform == 'win32':
                self.loop = asyncio.ProactorEventLoop()
            else:
                self.loop = asyncio.get_event_loop()
        else:
            self.loop = loop
        # Starting by add "tasks" in "loop"
        if self.mode == "Reader":
            self._start_reader()
        elif self.mode == "Run":
            self._start_run()
        else:
            raise TypeError("self.mode is not Reader or Run.")
        # Start or not loop.run_forever
        if self.run_loop:
            try:
                print("Cmd._start_controller start loop inside Cmd object!")
                self.stdout.flush()
                self.loop.run_forever()
            except KeyboardInterrupt:
                print("Cmd._start_controller stop loop. Bye.")
                self.loop.stop()
                pending = asyncio.Task.all_tasks(loop=self.loop)
                print(asyncio.Task.all_tasks(loop=self.loop))
                for task in pending:
                    task.cancel()
                    with suppress(asyncio.CancelledError):
                        self.loop.run_until_complete(task)
                # self.loop.close()

def main():
    """Main function"""
    global _ENABLE_P2P, _CONCURRENT_DOWNLOADS_ALLOWED, _POOL_ID
    # get command-line args
    args = parseargs()
    p2popts = args.p2popts.split(':')
    _ENABLE_P2P = p2popts[0] == 'true'
    _CONCURRENT_DOWNLOADS_ALLOWED = int(p2popts[1])
    logger.info('max concurrent downloads: {}'.format(
        _CONCURRENT_DOWNLOADS_ALLOWED))
    # set p2p options
    if _ENABLE_P2P:
        if not _LIBTORRENT_IMPORTED:
            raise ImportError('No module named \'libtorrent\'')
        global _COMPRESSION, _SEED_BIAS, _SAVELOAD_FILE_EXTENSION
        _COMPRESSION = p2popts[3] == 'true'
        _SEED_BIAS = int(p2popts[2])
        if not _COMPRESSION:
            _SAVELOAD_FILE_EXTENSION = 'tar'
        logger.info('peer-to-peer options: compression={} seedbias={}'.format(
            _COMPRESSION, _SEED_BIAS))
        # create torrent directory
        logger.debug('creating torrent dir: {}'.format(_TORRENT_DIR))
        _TORRENT_DIR.mkdir(parents=True, exist_ok=True)
    del p2popts

    # get event loop
    if _ON_WINDOWS:
        loop = asyncio.ProactorEventLoop()
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()
    loop.set_debug(True)

    # get ip address if not specified, for local testing only
    if args.ipaddress is None:
        ipaddress = loop.run_until_complete(_get_ipaddress_async(loop))
    else:
        ipaddress = args.ipaddress
    logger.debug('ip address: {}'.format(ipaddress))

    # set up storage names
    _setup_storage_names(args.prefix)
    del args

    # create storage credentials
    blob_client, table_client = _create_credentials()

    # distribute global resources
    distribute_global_resources(loop, blob_client, table_client, ipaddress)

def __init__(self, backends=None, middlewares=None, state=None,
                 process_types=None, extra={}, state_path=None,
                 storage_engine="local", conf=None,
                 config_path=None, encrypt=True, **kwargs):

        self.process_types = process_types
        self.extra = extra

        self.conf = conf or Config(data_path=state_path,
                                   config_path=config_path,
                                   encrypt=encrypt,
                                   **kwargs)

        self.conf.init()
        self.conf.init_logging()

        if self.conf.storage['encrypt']:
            self.fernet = self.conf.get_fernet_token()

        # this is passed to all middlewares
        self.kwargs = kwargs

        StorageCls = self.load_thing(self.possible_storages[storage_engine])

        self.storage = StorageCls(self, encrypt=encrypt, **self.kwargs)

        # state persistence
        try:
            self.state = self.storage.restore_state(self.conf.data_path)
            self.conf.init()
            self.conf.init_logging()
        except:
            self.state = StateMachine(self)
        finally:
            self.state.state_path = self.conf.data_path

        # load dynamic parts
        self.load_backends(backends or self.conf.core['backends'])

        self.executor = ThreadPoolExecutor(max_workers=EXECUTOR_POOL_SIZE)
        # initialize pools
        if sys.platform == 'win32':
            self.loop = asyncio.ProactorEventLoop()
        else:
            self.loop = asyncio.get_event_loop()

        self.loop.set_default_executor(self.executor)
        self.loop.set_exception_handler(self._async_exception_handler)

def get_test_home_assistant():
    """Return a Home Assistant object pointing at test config directory."""
    if sys.platform == "win32":
        loop = asyncio.ProactorEventLoop()
    else:
        loop = asyncio.new_event_loop()

    hass = loop.run_until_complete(async_test_home_assistant(loop))

    # FIXME should not be a daemon. Means hass.stop() not called in teardown
    stop_event = threading.Event()

    def run_loop():
        """Run event loop."""
        # pylint: disable=protected-access
        loop._thread_ident = threading.get_ident()
        loop.run_forever()
        loop.close()
        stop_event.set()

    threading.Thread(name="LoopThread", target=run_loop, daemon=True).start()

    orig_start = hass.start
    orig_stop = hass.stop

    @patch.object(hass.loop, 'run_forever')
    @patch.object(hass.loop, 'close')
    def start_hass(*mocks):
        """Helper to start hass."""
        orig_start()
        hass.block_till_done()

    def stop_hass():
        """Stop hass."""
        orig_stop()
        stop_event.wait()

    hass.start = start_hass
    hass.stop = stop_hass

    return hass


# pylint: disable=protected-access