我如何避免循环参数

问题描述

以下代码是我在python 3.6中编写的一些自动化测试的一部分:

The following code is part of some automated tests that I have written in python 3.6:

connected = False

def aiohttp_server(loop):
    async def handler(msg, session):
        global connected
        if msg.type == sockjs.MSG_OPEN:
            connected = True
        if msg.type == sockjs.MSG_CLOSE:
            connected = False

    app = web.Application(loop=loop)
    sockjs.add_endpoint(app, handler)
    runner = web.AppRunner(app)
    return runner

def run_server(runner, loop):
    logging.basicConfig(level=logging.DEBUG,
                        format='%(asctime)s %(levelname)s %(message)s')

    asyncio.set_event_loop(loop)
    loop.run_until_complete(runner.setup())
    site = web.TCPSite(runner, 'localhost', 8080)
    loop.run_until_complete(site.start())
    loop.run_forever()

def start_server():
    loop = asyncio.new_event_loop()
    t = threading.Thread(target=run_server, args=(aiohttp_server(loop),loop,), daemon=True)
    t.start()
    time.sleep(0.01)

基本上，调用start_server应该使用名为/sockjs

Basically, calling start_server should initiate a simple web server with a sockjs endpoint named /sockjs

我还不是python的async关键字的熟练者.我怀疑有两个相关的问题:

I am not yet a master of python's async keyword. There are two issues, that I suspect are related:

首先，我在app = web.Application(loop=loop)语句上收到了弃用警告:

Firstly, I am getting a deprecation warning on the app = web.Application(loop=loop) statement:

/home/peter/incubator/sockjs_client/tests/test_sockjs_client.py:25: DeprecationWarning: loop argument is deprecated
  app = web.Application(loop=loop)
/home/peter/.local/lib/python3.6/site-packages/sockjs/route.py:54: DeprecationWarning: loop property is deprecated
  manager = SessionManager(name, app, handler, app.loop)

其次，测试偶尔会失败.我相信，根据机器负载，有时服务器没有足够的时间来启动测试代码开始真正执行.

And secondly, the tests fail occasionally. I believe that, depending on machine load, sometimes the server hasn't had enough time to start before the test code actually starts executing.

基本上，我需要的是start_server函数用websocket终结点初始化Web应用程序，直到应用程序准备好接受Websocket连接时才返回.

Basically, what I need is for the start_server function to initialise a web application with a websocket endpoint, and not return until the application is prepared to accept websocket connections.

推荐答案

避免在任何地方绕过loop的推荐方法是切换到 asyncio.run .不用手动管理循环，而是让asyncio.run为您创建(并关闭)循环.如果您的所有工作都在协程中完成，则可以使用get_event_loop()或get_running_loop()访问循环.

The recommended way to avoid passing around the loop everywhere is to switch to asyncio.run. Instead of managing the loop manually, let asyncio.run create (and close) the loop for you. If all your work is done in coroutines, you can access the loop with get_event_loop() or get_running_loop().

您可以将threading.Event传递给在设置站点时设置的线程，并在主线程中等待它.

You can pass a threading.Event to the thread that gets set when the site is set up, and wait for it in the main thread.

这是一个(未试用的)示例，它同时实现了两个建议:

Here is an (untested) example that implements both suggestions:

connected = False

def aiohttp_server():
    async def handler(msg, session):
        global connected
        if msg.type == sockjs.MSG_OPEN:
            connected = True
        if msg.type == sockjs.MSG_CLOSE:
            connected = False

    app = web.Application()
    sockjs.add_endpoint(app, handler)
    return web.AppRunner(app)

async def run_server(ready):
    logging.basicConfig(level=logging.DEBUG,
                        format='%(asctime)s %(levelname)s %(message)s')
    runner = aiohttp_server()
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    ready.set()
    # emulates loop.run_forever()
    await asyncio.get_running_loop().create_future()

def start_server():
    ready = threading.Event()
    threading.Thread(target=asyncio.run, args=(aiohttp_server(ready),),
                     daemon=True).start()
    ready.wait()

这篇关于我如何避免循环参数的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！