aiohttp_gunicorn_worker_test.py

# Worker fails here if it closes, creates and uses the same loop
# for its runner and apps code it's serving.
#
# Start with: gunicorn -w 2 aiohttp_gunicorn_worker_test:app --worker-class aiohttp.worker.AsyncGunicornWorker

import asyncio
import aiohttp


@asyncio.coroutine
def get_page():
    r = yield from aiohttp.request('GET', 'http://example.com')
    r.close()
    yield from asyncio.sleep(1)


def app(environ, start_response):
    data = b"Hello, World!\n"
    start_response("200 OK", [
        ("Content-Type", "text/plain"),
        ("Content-Length", str(len(data)))
    ])

    # when creating and setting up new event loop, it works.
    # if loop is obtained from asyncio.get_event_loop() worker fails.
    loop = asyncio.get_event_loop()
    loop.run_until_complete(get_page())
    return iter([data])