def listen_message_stream(self, id_blacklist=None):
        id_blacklist = set(id_blacklist or [self.me, ])

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        with aiohttp.ClientSession(loop=loop) as session:
            self.aioclient_session = session

            tasks = [
                asyncio.ensure_future(self.fetch(session, room, id_blacklist))
                for room in self.rooms
            ]
            done, _ = loop.run_until_complete(
                asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
            )
            for d in done:
                if d.exception():
                    raise d.exception() 

def test_logo_base(app, caplog):
    server = app.create_server(
        debug=True, return_asyncio_server=True, port=PORT
    )
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop._stopping = False

    with caplog.at_level(logging.DEBUG):
        _server = loop.run_until_complete(server)

    _server.close()
    loop.run_until_complete(_server.wait_closed())
    app.stop()

    assert caplog.record_tuples[ROW][1] == logging.DEBUG
    assert caplog.record_tuples[ROW][2] == BASE_LOGO 

def test_logo_false(app, caplog):
    app.config.LOGO = False

    server = app.create_server(
        debug=True, return_asyncio_server=True, port=PORT
    )
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop._stopping = False

    with caplog.at_level(logging.DEBUG):
        _server = loop.run_until_complete(server)

    _server.close()
    loop.run_until_complete(_server.wait_closed())
    app.stop()

    banner, port = caplog.record_tuples[ROW][2].rsplit(":", 1)
    assert caplog.record_tuples[ROW][1] == logging.INFO
    assert banner == "Goin' Fast @ http://127.0.0.1"
    assert int(port) > 0 

def test_logo_custom(app, caplog):
    app.config.LOGO = "My Custom Logo"

    server = app.create_server(
        debug=True, return_asyncio_server=True, port=PORT
    )
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop._stopping = False

    with caplog.at_level(logging.DEBUG):
        _server = loop.run_until_complete(server)

    _server.close()
    loop.run_until_complete(_server.wait_closed())
    app.stop()

    assert caplog.record_tuples[ROW][1] == logging.DEBUG
    assert caplog.record_tuples[ROW][2] == "My Custom Logo" 

def test_keep_alive_timeout_reuse():
    """If the server keep-alive timeout and client keep-alive timeout are
     both longer than the delay, the client _and_ server will successfully
     reuse the existing connection."""
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        client = ReuseableSanicTestClient(keep_alive_timeout_app_reuse, loop)
        headers = {"Connection": "keep-alive"}
        request, response = client.get("/1", headers=headers)
        assert response.status == 200
        assert response.text == "OK"
        loop.run_until_complete(aio_sleep(1))
        request, response = client.get("/1")
        assert response.status == 200
        assert response.text == "OK"
    finally:
        client.kill_server() 

def test_keep_alive_client_timeout():
    """If the server keep-alive timeout is longer than the client
    keep-alive timeout, client will try to create a new connection here."""
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        client = ReuseableSanicTestClient(keep_alive_app_client_timeout, loop)
        headers = {"Connection": "keep-alive"}
        try:
            request, response = client.get(
                "/1", headers=headers, request_keepalive=1
            )
            assert response.status == 200
            assert response.text == "OK"
            loop.run_until_complete(aio_sleep(2))
            exception = None
            request, response = client.get("/1", request_keepalive=1)
        except ValueError as e:
            exception = e
        assert exception is not None
        assert isinstance(exception, ValueError)
        assert "got a new connection" in exception.args[0]
    finally:
        client.kill_server() 

def loop():
    """Creates new event loop."""
    loop = asyncio.new_event_loop()
    if sys.version_info < (3, 8):
        asyncio.set_event_loop(loop)

    try:
        yield loop
    finally:
        if hasattr(loop, 'is_closed'):
            closed = loop.is_closed()
        else:
            closed = loop._closed   # XXX
        if not closed:
            loop.call_soon(loop.stop)
            loop.run_forever()
            loop.close() 

def _run_in_fresh_loop(coro, timeout=30):
    thres = []
    thexc = []

    def run():
        loop = asyncio.new_event_loop()
        try:
            task = loop.create_task(coro(loop=loop))
            thres.append(loop.run_until_complete(task))
        except Exception as e:
            thexc.append(e)
        finally:
            loop.close()

    th = threading.Thread(target=run)
    th.start()
    th.join(timeout=timeout)

    # re-raise a thread exception
    if len(thexc) > 0:
        raise thexc[0]

    return thres[0] 

def run_worker(input):
    if os.getpid() not in _PROC:
        _PROC.append(os.getpid())
    _CONSOLE.print("hello")
    try:
        3 + ""
    except Exception:
        _CONSOLE.print_error("meh")

    with catch_output() as (stdout, stderr):
        loop = asyncio.new_event_loop()
        fut = asyncio.ensure_future(_CONSOLE.display(), loop=loop)
        loop.run_until_complete(fut)
        loop.close()

    stdout = stdout.read()
    assert stdout == "", stdout 

def async_test(func):
    @functools.wraps(func)
    def _async_test(*args, **kw):
        cofunc = asyncio.coroutine(func)
        oldloop = asyncio.get_event_loop()
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.set_debug(True)
        console = SharedConsole(interval=0)
        results = SharedCounters(
            "WORKER", "REACHED", "RATIO", "OK", "FAILED", "MINUTE_OK", "MINUTE_FAILED"
        )
        kw["loop"] = loop
        kw["console"] = console
        kw["results"] = results
        try:
            loop.run_until_complete(cofunc(*args, **kw))
        finally:
            loop.stop()
            loop.close()
            asyncio.set_event_loop(oldloop)

    return _async_test 

def ad_hoc() -> None:
    try:
        event_loop = asyncio.get_event_loop()
    except RuntimeError:
        event_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(event_loop)

    league.set_status(league.Status.CLOSED)
    multiverse.init() # New Cards?
    event_loop.run_until_complete(multiverse.set_legal_cards_async()) # PD current list
    event_loop.run_until_complete(multiverse.update_pd_legality_async()) # PD previous lists
    insert_seasons.run() # Make sure Season table is up to date
    if redis.REDIS: # Clear the redis cache
        redis.REDIS.flushdb()
    league_end = league.active_league().end_date
    diff = league_end - dtutil.now()
    if diff.days > 0:
        league.set_status(league.Status.OPEN)
    print('Open the gates here')
    reprime_cache.run() # Update deck legalities
    if redis.REDIS: # Clear the redis cache
        redis.REDIS.flushdb() 

def setUp(self):
        setup_mock_web_api_server(self)

        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        task = asyncio.ensure_future(self.mock_server(), loop=self.loop)
        self.loop.run_until_complete(asyncio.wait_for(task, 0.1))

        self.client = slack.RTMClient(
            token="xoxb-valid",
            base_url="http://localhost:8765",
            auto_reconnect=False,
            run_async=False,
        )
        self.client._web_client = slack.WebClient(
            token="xoxb-valid",
            base_url="http://localhost:8888",
            run_async=False,
        ) 

def test_the_cost_of_event_loop_creation(self):
        # create 100 event loops
        loops = []
        try:
            upper_limit = 0
            for i in range(1000):
                try:
                    loops.append(asyncio.new_event_loop())
                except OSError as e:
                    self.logger.info(f"Got an OSError when creating {i} event loops")
                    self.assertEqual(e.errno, 24)
                    self.assertEqual(e.strerror, "Too many open files")
                    upper_limit = i
                    break
            self.assertTrue(upper_limit > 0)
        finally:
            for loop in loops:
                loop.close() 

def per_request_async():
    try:
        # This is not optimal and the host should have a large number of FD (File Descriptor)
        loop_for_this_request = asyncio.new_event_loop()

        async_client = WebClient(
            token=os.environ["SLACK_BOT_TOKEN"],
            run_async=True,
            loop=loop_for_this_request
        )
        future = async_client.chat_postMessage(
            channel="#random",
            text="You used the singleton WebClient for posting this message!"
        )
        response = loop_for_this_request.run_until_complete(future)
        return str(response)
    except SlackApiError as e:
        return make_response(str(e), 400) 

def handle(self, *args, **kwargs):
        """Run the executor listener. This method never returns."""
        listener = ExecutorListener(
            redis_params=getattr(settings, "FLOW_MANAGER", {}).get(
                "REDIS_CONNECTION", {}
            )
        )

        def _killer(signum, frame):
            """Kill the listener on receipt of a signal."""
            listener.terminate()

        signal(SIGINT, _killer)
        signal(SIGTERM, _killer)

        async def _runner():
            """Run the listener instance."""
            if kwargs["clear_queue"]:
                await listener.clear_queue()
            async with listener:
                pass

        loop = asyncio.new_event_loop()
        loop.run_until_complete(_runner())
        loop.close() 

def _run_in_event_loop(coro, *args, **kwargs):
    """Run a coroutine in a runloop call.

    This is needed as the top level call into Resolwe Manager-using
    tests. An event loop is started so that it can be used within the
    call tree.

    :param coro: The coroutine to run with an underlying event loop. All
        other arguments given to this function are forwarded to it.
    """
    asyncio.get_event_loop().close()
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    task = asyncio.ensure_future(coro(*args, **kwargs), loop=loop)
    loop.run_until_complete(task)
    loop.close()
    return task.result() 

def start_node():
    config = configure_node()
    loop = asyncio.new_event_loop()
    node = tattle.Cluster(config, loop=loop)
    thread = NodeThread(node, loop=loop)
    thread.start()

    other_node = random.choice(list(all_nodes)) if all_nodes else None

    all_nodes.add(node)
    all_threads[node] = thread
    client = NodeClient(port=config.api_port)

    if other_node:
        client.join((other_node.local_node_address, other_node.local_node_port))

    return client 

def _restart_stream(self, stream_id):
        """
        This is NOT stop/start! Its purpose is to start a died stream again! Use `set_restart_request()` for stop/start!

        :param stream_id: id of a stream
        :type stream_id: uuid

        :return: stream_id or False
        """
        try:
            if self.restart_requests[stream_id]['status'] != "new":
                logging.warning("BinanceWebSocketApiManager->_restart_stream() please use `set_restart_request() "
                                "instead!")
                return False
        except KeyError:
            # no restart_request entry for this stream_id:
            logging.warning("BinanceWebSocketApiManager->_restart_stream() please use `set_restart_request() instead!")
            return False
        logging.info("BinanceWebSocketApiManager->_restart_stream(" + str(stream_id) + ", " +
                     str(self.stream_list[stream_id]['channels']) +
                     ", " + str(self.stream_list[stream_id]['markets']) + ")")
        self.restart_requests[stream_id] = {'status': "restarted"}
        self.restart_requests[stream_id]['last_restart_time'] = time.time()
        self.stream_list[stream_id]['status'] = "restarting"
        self.stream_list[stream_id]['kill_request'] = None
        self.stream_list[stream_id]['payload'] = []
        loop = asyncio.new_event_loop()
        thread = threading.Thread(target=self._create_stream_thread, args=(loop, stream_id,
                                                                           self.stream_list[stream_id]['channels'],
                                                                           self.stream_list[stream_id]['markets'],
                                                                           self.stream_list[stream_id]['stream_label'],
                                                                           self.stream_list[stream_id]['stream_buffer_name'],
                                                                           True))
        thread.start()
        return stream_id 

def run(self):
        self.loop = loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        try:
            loop.run_forever()
        finally:
            try:
                shutdown_asyncgens = loop.shutdown_asyncgens()
            except AttributeError:
                pass
            else:
                loop.run_until_complete(shutdown_asyncgens)
            loop.close()
            asyncio.set_event_loop(None) 

def test_worker_init_process(worker):
    with mock.patch("sanic.worker.asyncio") as mock_asyncio:
        try:
            worker.init_process()
        except TypeError:
            pass

        assert mock_asyncio.get_event_loop.return_value.close.called
        assert mock_asyncio.new_event_loop.called
        assert mock_asyncio.set_event_loop.called 

def test_run_max_requests_exceeded(worker):
    loop = asyncio.new_event_loop()
    worker.ppid = 1
    worker.alive = True
    sock = mock.Mock()
    sock.cfg_addr = ("localhost", 8080)
    worker.sockets = [sock]
    worker.wsgi = mock.Mock()
    worker.connections = set()
    worker.log = mock.Mock()
    worker.loop = loop
    worker.servers = {
        "server1": {"requests_count": 14},
        "server2": {"requests_count": 15},
    }
    worker.max_requests = 10
    worker._run = mock.Mock(wraps=_a_noop)

    # exceeding request count
    _runner = asyncio.ensure_future(worker._check_alive(), loop=loop)
    loop.run_until_complete(_runner)

    assert not worker.alive
    worker.notify.assert_called_with()
    worker.log.info.assert_called_with(
        "Max requests exceeded, shutting " "down: %s", worker
    ) 

def test_worker_close(worker):
    loop = asyncio.new_event_loop()
    asyncio.sleep = mock.Mock(wraps=_a_noop)
    worker.ppid = 1
    worker.pid = 2
    worker.cfg.graceful_timeout = 1.0
    worker.signal = mock.Mock()
    worker.signal.stopped = False
    worker.wsgi = mock.Mock()
    conn = mock.Mock()
    conn.websocket = mock.Mock()
    conn.websocket.close_connection = mock.Mock(wraps=_a_noop)
    worker.connections = set([conn])
    worker.log = mock.Mock()
    worker.loop = loop
    server = mock.Mock()
    server.close = mock.Mock(wraps=lambda *a, **kw: None)
    server.wait_closed = mock.Mock(wraps=_a_noop)
    worker.servers = {server: {"requests_count": 14}}
    worker.max_requests = 10

    # close worker
    _close = asyncio.ensure_future(worker.close(), loop=loop)
    loop.run_until_complete(_close)

    assert worker.signal.stopped
    assert conn.websocket.close_connection.called
    assert len(worker.servers) == 0 

def test_keep_alive_server_timeout():
    """If the client keep-alive timeout is longer than the server
    keep-alive timeout, the client will either a 'Connection reset' error
    _or_ a new connection. Depending on how the event-loop handles the
    broken server connection."""
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        client = ReuseableSanicTestClient(keep_alive_app_server_timeout, loop)
        headers = {"Connection": "keep-alive"}
        try:
            request, response = client.get(
                "/1", headers=headers, request_keepalive=60
            )
            assert response.status == 200
            assert response.text == "OK"
            loop.run_until_complete(aio_sleep(3))
            exception = None
            request, response = client.get("/1", request_keepalive=60)
        except ValueError as e:
            exception = e
        assert exception is not None
        assert isinstance(exception, ValueError)
        assert (
            "Connection reset" in exception.args[0]
            or "got a new connection" in exception.args[0]
        )
    finally:
        client.kill_server() 

def test_windows_workaround():
    """Test Windows workaround (on any other OS)"""
    # At least some code coverage, even though this test doesn't work on
    # Windows...
    class MockApp:
        def __init__(self):
            self.is_stopping = False

        def stop(self):
            assert not self.is_stopping
            self.is_stopping = True

        def add_task(self, func):
            loop = asyncio.get_event_loop()
            self.stay_active_task = loop.create_task(func(self))

    async def atest(stop_first):
        app = MockApp()
        ctrlc_workaround_for_windows(app)
        await asyncio.sleep(0.05)
        if stop_first:
            app.stop()
            await asyncio.sleep(0.2)
        assert app.is_stopping == stop_first
        # First Ctrl+C: should call app.stop() within 0.1 seconds
        os.kill(os.getpid(), signal.SIGINT)
        await asyncio.sleep(0.2)
        assert app.is_stopping
        assert app.stay_active_task.result() == None
        # Second Ctrl+C should raise
        with pytest.raises(KeyboardInterrupt):
            os.kill(os.getpid(), signal.SIGINT)
        return "OK"

    # Run in our private loop
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    res = loop.run_until_complete(atest(False))
    assert res == "OK"
    res = loop.run_until_complete(atest(True))
    assert res == "OK" 

def loop(request):
    asyncio.set_event_loop(None)
    loop = asyncio.new_event_loop()

    def fin():
        loop.close()

    request.addfinalizer(fin)
    return loop 

def loop(request):
    asyncio.set_event_loop(None)
    loop = asyncio.new_event_loop()

    def fin():
        loop.close()

    request.addfinalizer(fin)
    return loop 

def run(cls):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        plugin = cls(loop)
        try:
            loop.create_task(plugin.install())
            loop.run_forever()
        except KeyboardInterrupt:
            pass
        except Exception as e:
            logger.error('run failed: %s', repr(e), exc_info=True)
        finally:
            loop.run_until_complete(plugin.uninstall())
        loop.close() 

def __init__(self):
        self._loop = asyncio.new_event_loop()
        self._thread = threading.Thread(target=self._target)
        self._thread.start() 

def initialize(self, **kwargs):
        loop = asyncio.new_event_loop()
        try:
            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)
        except Exception:
            # If initialize() does not succeed (taking ownership of the loop),
            # we have to close it.
            loop.close()
            raise 

def initialize(self, **kwargs):
        loop = asyncio.new_event_loop()
        try:
            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)
        except Exception:
            # If initialize() does not succeed (taking ownership of the loop),
            # we have to close it.
            loop.close()
            raise