def main():
    logger.info('VLC Scheduler v%s started.' % version.VERSION)
    
    if sys.platform == 'win32':
        loop = asyncio.ProactorEventLoop()
        asyncio.set_event_loop(loop)
    
    loop = asyncio.get_event_loop()
    
    try:
        loop.run_until_complete(main_coro())
    except Exception as e:
        if config.DEBUG:
            logger.fatal(traceback.format_exc())
        else:
            logger.fatal(str(e))
    finally:
        loop.close()
        logger.info('VLC Scheduler stopped.') 

def parse_arguments():
    """Parse the arguments"""
    parser = argparse.ArgumentParser()
    parser.add_argument("--setup", "-s", action="store_true")
    parser.add_argument("--phone", "-p", action="append")
    parser.add_argument("--token", "-t", action="append", dest="tokens")
    parser.add_argument("--heroku", action="store_true")
    parser.add_argument("--local-db", dest="local", action="store_true")
    parser.add_argument("--web-only", dest="web_only", action="store_true")
    parser.add_argument("--no-web", dest="web", action="store_false")
    parser.add_argument("--heroku-web-internal", dest="heroku_web_internal", action="store_true",
                        help="This is for internal use only. If you use it, things will go wrong.")
    arguments = parser.parse_args()
    logging.debug(arguments)
    if sys.platform == "win32":
        # Subprocess support; not needed in 3.8 but not harmful
        asyncio.set_event_loop(asyncio.ProactorEventLoop())

    return arguments 

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_true(app, caplog):
    app.config.LOGO = True

    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_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 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_via_asyncio(async_to_run, debug_event_loop=False):
    logger = logging.getLogger('asyncio.main')

    asyncio.set_event_loop(asyncio.new_event_loop())
    event_loop = asyncio.get_event_loop()
    event_loop.set_debug(enabled=debug_event_loop)
    try:
        logger.info("starting events loop ...")
        event_loop.run_until_complete(async_to_run)

        _cleanup_remaining_tasks(loop=event_loop, logger=logger)

    finally:
        logger.info("closing events loop ...")
        event_loop.close()
        logger.info("... events loop closed") 

def run_via_asyncio(async_to_run, debug_event_loop=False):
    logger = logging.getLogger('asyncio.main')

    asyncio.set_event_loop(asyncio.new_event_loop())
    event_loop = asyncio.get_event_loop()
    event_loop.set_debug(enabled=debug_event_loop)
    try:
        logger.info("starting events loop ...")
        event_loop.run_until_complete(async_to_run)

        _cleanup_remaining_tasks(loop=event_loop, logger=logger)

    finally:
        logger.info("closing events loop ...")
        event_loop.close()
        logger.info("... events loop closed")


# configure library directly from dict 

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 dedicatedloop_noclose(func):
    @functools.wraps(func)
    def _loop(*args, **kw):
        old_loop = asyncio.get_event_loop()
        loop = asyncio.new_event_loop()
        loop.set_debug(True)
        loop._close = loop.close
        loop.close = lambda: None
        asyncio.set_event_loop(loop)
        try:
            return func(*args, **kw)
        finally:
            loop._close()
            asyncio.set_event_loop(old_loop)

    return _loop 

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 _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 _create_stream_thread(self, loop, stream_id, channels, markets, stream_label=None, stream_buffer_name=False,
                              restart=False):
        """
        Co function of self.create_stream to create a thread for the socket and to manage the coroutine

        :param loop: provide a asynio loop
        :type loop: asyncio loop
        :param stream_id: provide a stream_id (only needed for userData Streams (acquiring a listenKey)
        :type stream_id: uuid
        :param channels: provide the channels to create the URI
        :type channels: str, tuple, list, set
        :param markets: provide the markets to create the URI
        :type markets: str, tuple, list, set
        :param stream_label: provide a stream_label for the stream
        :type stream_label: str
        :param stream_buffer_name: If `False` the data is going to get written to the default stream_buffer,
                           set to `True` to read the data via `pop_stream_data_from_stream_buffer(stream_id)` or
                           provide a string to create and use a shared stream_buffer and read it via
                           `pop_stream_data_from_stream_buffer('string')`.
        :type stream_buffer_name: bool or str
        :param restart: set to `True`, if its a restart!
        :type restart: bool
        :return:
        """
        if self.is_stop_request(stream_id):
            return False
        if restart is False:
            self._add_socket_to_socket_list(stream_id, channels, markets, stream_label, stream_buffer_name)
            if stream_buffer_name is not False:
                self.stream_buffer_locks[stream_buffer_name] = threading.Lock()
                self.stream_buffers[stream_buffer_name] = []
        asyncio.set_event_loop(loop)
        binance_websocket_api_socket = BinanceWebSocketApiSocket(self, stream_id, channels, markets)
        try:
            loop.run_until_complete(binance_websocket_api_socket.start_socket())
        finally:
            loop.close() 

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_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 init_process(self):
        # create new event_loop after fork
        asyncio.get_event_loop().close()

        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

        super().init_process() 

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 test_consul_ctor(self, loop, consul_port):
        # same as previous but with global event loop
        async def main():
            c = consul.aio.Consul(port=consul_port)
            assert c._loop is loop
            await c.kv.put('foo', struct.pack('i', 1000))
            index, data = await c.kv.get('foo')
            assert struct.unpack('i', data['Value']) == (1000,)

        asyncio.set_event_loop(loop)
        loop.run_until_complete(main()) 

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 _target(self):
        asyncio.set_event_loop(self._loop)
        self._loop.run_forever() 

def run(self):
        #self.loop = asyncio.new_event_loop()
        #asyncio.set_event_loop(self.loop)
        self.loop = asyncio.get_event_loop()
        self.loop.run_until_complete(self.asyncio_loop()) 

def test_global_loop():
    conn = mock.Mock(spec=(
        'execute closed _transaction_error _buffered'
        .split()))
    try:
        old_loop = asyncio.get_event_loop()
    except (AssertionError, RuntimeError):
        old_loop = None
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    tr = MultiExec(conn, commands_factory=Redis)
    # assert tr._loop is loop

    def make_fut(cmd, *args, **kw):
        fut = asyncio.get_event_loop().create_future()
        if cmd == 'PING':
            fut.set_result(b'QUEUED')
        elif cmd == 'EXEC':
            fut.set_result([b'PONG'])
        else:
            fut.set_result(b'OK')
        return fut

    conn.execute.side_effect = make_fut
    conn.closed = False
    conn._transaction_error = None
    conn._buffered.side_effect = lambda: nullcontext(conn)

    async def go():
        tr.ping()
        res = await tr.execute()
        assert res == [b'PONG']
    loop.run_until_complete(go())
    asyncio.set_event_loop(old_loop) 

def compat_event_loop():
    """OS agnostic context manager for an event loop."""
    if sys.platform.startswith("win"):
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

    event_loop = asyncio.get_event_loop()

    if event_loop.is_closed():
        event_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(event_loop)

    yield event_loop

    event_loop.close() 

def thread_secure_get_event_loop(logger_name="moler.runner.asyncio"):
    """
    Need securing since asyncio.get_event_loop() when called from new thread
    may raise sthg like:
    RuntimeError: There is no current event loop in thread 'Thread-3'
    It is so since MainThread has preinstalled loop but other threads must
    setup own loop by themselves.

    :return: loop of current thread + info if it was newly created
    """
    new_loop = False
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError as err:
        if "no current event loop in thread" in str(err):
            loop = asyncio.new_event_loop()
            loop_id = instance_id(loop)
            logger = logging.getLogger(logger_name)
            logger.debug("created new event loop {}:{}".format(loop_id, loop))
            asyncio.set_event_loop(loop)
            new_loop = True
        else:
            raise

    loop.set_debug(enabled=True)
    return loop, new_loop