def run_sync(func: Callable[..., Any]) -> Callable[..., Coroutine[Any, None, None]]:
    """Ensure that the sync function is run within the event loop.

    If the *func* is not a coroutine it will be wrapped such that
    it runs in the default executor (use loop.set_default_executor
    to change). This ensures that synchronous functions do not
    block the event loop.
    """

    @wraps(func)
    async def _wrapper(*args: Any, **kwargs: Any) -> Any:
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(
            None, copy_context().run, partial(func, *args, **kwargs)
        )
        if isgenerator(result):
            return run_sync_iterable(result)  # type: ignore
        else:
            return result

    _wrapper._quart_async_wrapper = True  # type: ignore
    return _wrapper 

def run_sync_iterable(iterable: Generator[Any, None, None]) -> AsyncGenerator[Any, None]:
    async def _gen_wrapper() -> AsyncGenerator[Any, None]:
        # Wrap the generator such that each iteration runs
        # in the executor. Then rationalise the raised
        # errors so that it ends.
        def _inner() -> Any:
            # https://bugs.python.org/issue26221
            # StopIteration errors are swallowed by the
            # run_in_exector method
            try:
                return next(iterable)
            except StopIteration:
                raise StopAsyncIteration()

        loop = asyncio.get_running_loop()
        while True:
            try:
                yield await loop.run_in_executor(None, copy_context().run, _inner)
            except StopAsyncIteration:
                return

    return _gen_wrapper() 

def main():
    redis = await aioredis.create_redis_pool('redis://localhost')

    ch, = await redis.psubscribe('channel:*')
    assert isinstance(ch, aioredis.Channel)

    async def reader(channel):
        async for ch, message in channel.iter():
            print("Got message in channel:", ch, ":", message)
    asyncio.get_running_loop().create_task(reader(ch))

    await redis.publish('channel:1', 'Hello')
    await redis.publish('channel:2', 'World')

    redis.close()
    await redis.wait_closed() 

def main():
    redis = await aioredis.create_redis_pool('redis://localhost')

    ch1, ch2 = await redis.subscribe('channel:1', 'channel:2')
    assert isinstance(ch1, aioredis.Channel)
    assert isinstance(ch2, aioredis.Channel)

    async def reader(channel):
        async for message in channel.iter():
            print("Got message:", message)
    asyncio.get_running_loop().create_task(reader(ch1))
    asyncio.get_running_loop().create_task(reader(ch2))

    await redis.publish('channel:1', 'Hello')
    await redis.publish('channel:2', 'World')

    redis.close()
    await redis.wait_closed() 

def __aenter__(self):
        if self._entered:
            raise RuntimeError(
                f"TaskGroup {self!r} has been already entered")
        self._entered = True

        if self._loop is None:
            self._loop = asyncio.get_running_loop()

        self._parent_task = asyncio.current_task(self._loop)
        if self._parent_task is None:
            raise RuntimeError(
                f'TaskGroup {self!r} cannot determine the parent task')
        self._patch_task(self._parent_task)

        return self 

def connect(self, start: int, headers: bytes) -> int:
        added = 0
        bail = False
        loop = asyncio.get_running_loop()
        async with self._header_connect_lock:
            for height, chunk in self._iterate_chunks(start, headers):
                try:
                    # validate_chunk() is CPU bound and reads previous chunks from file system
                    await loop.run_in_executor(None, self.validate_chunk, height, chunk)
                except InvalidHeader as e:
                    bail = True
                    chunk = chunk[:(height-e.height)*self.header_size]
                written = 0
                if chunk:
                    self.io.seek(height * self.header_size, os.SEEK_SET)
                    written = self.io.write(chunk) // self.header_size
                    self.io.truncate()
                    # .seek()/.write()/.truncate() might also .flush() when needed
                    # the goal here is mainly to ensure we're definitely flush()'ing
                    await loop.run_in_executor(None, self.io.flush)
                    self._size = None
                added += written
                if bail:
                    break
        return added 

def get_auth_data_login(self, cookie_jar, credentials):
        code = parse_qs(urlparse(credentials['end_uri']).query)["code"][0]
        loop = asyncio.get_running_loop()

        s = requests.Session()
        url = f"{self.blizzard_oauth_url}/token"
        data = {
            "grant_type": "authorization_code",
            "redirect_uri": REDIRECT_URI,
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "code": code
        }
        response = await loop.run_in_executor(None, partial(s.post, url, data=data))
        response.raise_for_status()
        result = response.json()
        access_token = result["access_token"]
        self.auth_data = WebsiteAuthData(cookie_jar=cookie_jar, access_token=access_token, region=self.region)
        return self.auth_data

    # NOTE: use user data to present usertag/name to Galaxy, if this token expires and plugin cannot refresh it
    # use stored usertag/name if token validation fails, this is temporary solution, as we do not need that
    # endpoint for nothing else at this moment 

def simulator(request: requests.Request):
    token = request.headers.get("Authorization")
    if token:
        token = token[6:]  # Drop 'token '

    data = SimulatorSchema(await request.json())
    if data["pull_request"]:
        loop = asyncio.get_running_loop()
        title, summary = await loop.run_in_executor(
            None,
            functools.partial(
                _sync_simulator,
                data["mergify.yml"]["pull_request_rules"],
                *data["pull_request"],
                token=token,
            ),
        )
    else:
        title, summary = ("The configuration is valid", None)

    return responses.JSONResponse(
        status_code=200, content={"title": title, "summary": summary}
    ) 

def __init__(
        self, session, host, port, ssl_context, site, callback, is_unifi_os=False,
    ):
        """Create resources for websocket communication."""
        self.session = session
        self.ssl_context = ssl_context
        self.session_handler_callback = callback

        if is_unifi_os:
            self.url = f"wss://{host}:{port}/proxy/network/wss/s/{site}/events"
        else:
            self.url = f"wss://{host}:{port}/wss/s/{site}/events"

        self._loop = asyncio.get_running_loop()

        self._data = None
        self._state = None 

def start_async_server(self):
        try:
            xmppserverlog.info(
                "Starting XMPP Server at {}:{}".format(self.address[0], self.address[1])
            )

            loop = asyncio.get_running_loop()

            self.server = await loop.create_server(
                self.xmpp_protocol, host=self.address[0], port=self.address[1]
            )

            self.server_coro = loop.create_task(self.server.serve_forever())

        except PermissionError as e:
            xmppserverlog.error(e.strerror)
            asyncio.create_task(bumper.shutdown())
            pass

        except asyncio.CancelledError:
            pass

        except Exception as e:
            xmppserverlog.exception("{}".format(e))
            asyncio.create_task(bumper.shutdown()) 

def _signal_handling(logger: logging.Logger, client: naz.Client) -> None:
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.get_event_loop()

    try:
        for _signal in [signal.SIGHUP, signal.SIGQUIT, signal.SIGTERM]:
            loop.add_signal_handler(
                _signal,
                functools.partial(
                    asyncio.ensure_future,
                    _handle_termination_signal(logger=logger, _signal=_signal, client=client),
                ),
            )
    except ValueError as e:
        logger.log(
            logging.DEBUG,
            {
                "event": "naz.cli.signals",
                "stage": "end",
                "state": "this OS does not support the said signal",
                "error": str(e),
            },
        ) 

def async_turn_off(self):
        _LOGGER.debug("Light " + self._friendly_name + " turning off.")
        url = 'https://api.wyzecam.com/app/v2/device/set_property'

        payload = {
            'phone_id': self._api._device_id,
            'access_token': self._api._access_token,
            'device_model': self._device_model,
            'ts': '1575948896791',
            'sc': '01dd431d098546f9baf5233724fa2ee2',
            'sv': '107693eb44244a948901572ddab807eb',
            'device_mac': self._device_mac,
            'pvalue': "0",
            'pid': 'P3',
            'app_ver': 'com.hualai.WyzeCam___2.6.62'
        }

        loop = asyncio.get_running_loop()
        loop.create_task(self._api.async_do_request(url, payload))

        self._state = False
        self._just_changed_state = True 

def async_turn_on(self):
        _LOGGER.debug("Switch " + self._friendly_name + " turning on.")
        url = 'https://api.wyzecam.com/app/v2/device/set_property'

        payload = {
            'phone_id': self._api._device_id,
            'access_token': self._api._access_token,
            'device_model': self._device_model,
            'ts': '1575948896791',
            'sc': '01dd431d098546f9baf5233724fa2ee2',
            'sv': '107693eb44244a948901572ddab807eb',
            'device_mac': self._device_mac,
            'pvalue': "1",
            'pid': 'P3',
            'app_ver': 'com.hualai.WyzeCam___2.6.62'
        }

        loop = asyncio.get_running_loop()
        loop.create_task(self._api.async_do_request(url, payload))

        self._state = True
        self._just_changed_state = True 

def start(self):
        self.log.info("Starting Daemon Server")

        def master_close_cb():
            asyncio.ensure_future(self.stop())

        try:
            asyncio.get_running_loop().add_signal_handler(
                signal.SIGINT, master_close_cb
            )
            asyncio.get_running_loop().add_signal_handler(
                signal.SIGTERM, master_close_cb
            )
        except NotImplementedError:
            self.log.info("Not implemented")

        self.websocket_server = await websockets.serve(
            self.safe_handle, "localhost", 55400
        )

        self.log.info("Waiting Daemon WebSocketServer closure")
        print("Daemon server started", flush=True)
        await self.websocket_server.wait_closed()
        self.log.info("Daemon WebSocketServer closed") 

def pre_validate_blocks_multiprocessing(
        self, blocks: List[FullBlock]
    ) -> List[Tuple[bool, Optional[bytes32]]]:
        futures = []
        # Pool of workers to validate blocks concurrently
        for block in blocks:
            if self._shut_down:
                return [(False, None) for _ in range(len(blocks))]
            futures.append(
                asyncio.get_running_loop().run_in_executor(
                    self.pool,
                    pre_validate_finished_block_header,
                    self.constants,
                    bytes(block),
                )
            )
        results = await asyncio.gather(*futures)

        for i, (val, pos) in enumerate(results):
            if pos is not None:
                pos = bytes32(pos)
            results[i] = val, pos
        return results 

def start(self):
        self.log.info("Starting Websocket Server")

        def master_close_cb():
            asyncio.ensure_future(self.stop())

        try:
            asyncio.get_running_loop().add_signal_handler(
                signal.SIGINT, master_close_cb
            )
            asyncio.get_running_loop().add_signal_handler(
                signal.SIGTERM, master_close_cb
            )
        except NotImplementedError:
            self.log.info("Not implemented")

        await self.start_wallet()

        await self.connect_to_daemon()
        self.log.info("webSocketServer closed") 

def send_request(self, method, params, sensitive_params):
        """
        Send request

        :param method:
        :param params:
        :param sensitive_params: list of parameters that are anonymized before logging; \
            if False - no params are considered sensitive, if True - all params are considered sensitive
        """
        self._last_request_id += 1
        request_id = str(self._last_request_id)

        loop = asyncio.get_running_loop()
        future = loop.create_future()
        self._requests_futures[self._last_request_id] = (future, sensitive_params)

        logging.info(
            "Sending request: id=%s, method=%s, params=%s",
            request_id, method, anonymise_sensitive_params(params, sensitive_params)
        )

        self._send_request(request_id, method, params)
        return await future 

def _verify_ffmpeg_installed(self):
        if self._ffmpeg_installed:
            return
        self._ffmpeg_installed = False
        path = self._conf.ffmpeg_path
        if hasattr(self._conf, "data_dir"):
            path += os.path.pathsep + os.path.join(getattr(self._conf, "data_dir"), "ffmpeg", "bin")
        path += os.path.pathsep + self._env_copy.get("PATH", "")
        self._which_ffmpeg, self._which_ffprobe = await asyncio.get_running_loop().run_in_executor(
            None, self._which_ffmpeg_and_ffmprobe, path
        )
        if not self._which_ffmpeg:
            log.warning("Unable to locate ffmpeg executable. Path: %s", path)
            raise FileNotFoundError(f"Unable to locate ffmpeg executable. Path: {path}")
        if not self._which_ffprobe:
            log.warning("Unable to locate ffprobe executable. Path: %s", path)
            raise FileNotFoundError(f"Unable to locate ffprobe executable. Path: {path}")
        if os.path.dirname(self._which_ffmpeg) != os.path.dirname(self._which_ffprobe):
            log.warning("ffmpeg and ffprobe are in different folders!")

        await self._verify_executables()
        self._ffmpeg_installed = True 

def async_timed_cache(duration: int):
    def wrapper(func):
        cache: typing.Dict[typing.Tuple,
                           typing.Tuple[typing.Any, float]] = {}

        @functools.wraps(func)
        async def _inner(*args, **kwargs) -> typing.Any:
            loop = asyncio.get_running_loop()
            time_now = loop.time()
            key = tuple([args, tuple([tuple([k, kwargs[k]]) for k in kwargs])])
            if key in cache and (time_now - cache[key][1] < duration):
                return cache[key][0]
            to_cache = await func(*args, **kwargs)
            cache[key] = to_cache, time_now
            return to_cache
        return _inner
    return wrapper 

def communicate_async(self, server=None, sync=True, timeout=1.0):
        (
            success_pattern,
            failure_pattern,
            requestable,
        ) = self._get_response_patterns_and_requestable(server)
        if self._handle_async(sync, server):
            return
        loop = asyncio.get_running_loop()
        self._response_future = loop.create_future()
        server.osc_protocol.register(
            pattern=success_pattern,
            failure_pattern=failure_pattern,
            procedure=self._set_response_async,
            once=True,
        )
        server.send(requestable.to_osc())
        await asyncio.wait_for(self._response_future, timeout=timeout)
        return self._response 

def boot(
        self, port=DEFAULT_PORT, *, scsynth_path=None, options=None, **kwargs
    ):
        if self._is_running:
            raise supriya.exceptions.ServerOnline
        port = port or DEFAULT_PORT
        loop = asyncio.get_running_loop()
        self._boot_future = loop.create_future()
        self._quit_future = loop.create_future()
        self._options = new(options or Options(), **kwargs)
        scsynth_path = scsynth.find(scsynth_path)
        self._process_protocol = AsyncProcessProtocol()
        await self._process_protocol.boot(self._options, scsynth_path, port)
        if not await self._process_protocol.boot_future:
            self._boot_future.set_result(False)
            self._quit_future.set_result(True)
            raise supriya.exceptions.ServerCannotBoot
        self._ip_address = "127.0.0.1"
        self._is_owner = True
        self._port = port
        await self._connect()
        return self 

def loop_is_running() -> bool:
    """
    Determine if there is a running asyncio event loop.

    This helps enable "call this when event loop is running" logic (see: Twisted's `callWhenRunning`),
    which is currently not provided by asyncio.
    """
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return False
    return True 

def worker_connect(sockname):
    loop = asyncio.get_running_loop()
    waiter = loop.create_future()
    con = WorkerConnection(loop)
    tr, pr = await loop.create_unix_connection(
        lambda: WorkerProtocol(loop=loop, con_waiter=waiter, con=con),
        path=sockname)
    con._protocol = pr
    con._transport = tr
    await waiter
    return con 

def create_manager(*, runstate_dir: str, name: str,
                         worker_cls: type, worker_args: dict,
                         pool_size: int) -> Manager:

    loop = asyncio.get_running_loop()
    pool = Manager(
        loop=loop,
        runstate_dir=runstate_dir,
        worker_cls=worker_cls,
        worker_args=worker_args,
        name=name,
        pool_size=pool_size)

    await pool.start()
    return pool 

def create(cls, *, name: str=None):
        loop = asyncio.get_running_loop()
        return cls(_loop=loop, _name=name, _private=True) 

def getCompilerPool():
    loop = asyncio.get_running_loop()
    pool = getattr(loop, "__FG_compiler_pool", None)
    if pool is None:
        pool = CompilerPool()
        loop.__FG_compiler_pool = pool
    return pool 

def __init__(self, maxWorkers=5):
        self.loop = asyncio.get_running_loop()
        self.maxWorkers = maxWorkers
        self.workers = []
        self.availableWorkers = asyncio.Queue() 

def write(self, text):
        attrString = AppKit.NSAttributedString.alloc().initWithString_attributes_(text, self.textAttributes)
        st = self._textView.textStorage()
        st.appendAttributedString_(attrString)
        # If we call scrollToEnd right away it seems to have no effect.
        # If we defer to the next opportunity in the event loop it works fine.
        loop = asyncio.get_running_loop()
        loop.call_soon(self.scrollToEnd) 

def open_battlenet_browser(self):
        url = self.authentication_client.blizzard_battlenet_download_url
        log.info(f'Opening battle.net website: {url}')
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, lambda x: webbrowser.open(x, autoraise=True), url) 

def get_translation(self, target_language, query_string, tkk=''):
        self.result = ''
        self.target_language = target_language
        self.query_string = query_string
        tk = Token(tkk).calculate_token(self.query_string)
        if len(self.query_string) > 5000:
            return '(╯‵□′)╯︵┻━┻: Maximum characters exceeded...'
        parse_query = urllib.parse.quote_plus(self.query_string)
        url = self.get_url(self.target_language, parse_query, tk)
        url_alt = self.get_url(self.alternative_language, parse_query, tk)
        try:
            loop = asyncio.get_running_loop()
            resp = loop.run_in_executor(None, partial(self.get_resp, url))
            resp_alt = loop.run_in_executor(None, partial(self.get_resp, url_alt))
            [resp, resp_alt] = await asyncio.gather(resp, resp_alt)
            if resp[2] == self.target_language:
                self.result += f'^_^: Translate {resp[2]} To {self.alternative_language}\n'
                self.get_result(resp)
                self.get_result(resp_alt)
                self.get_synonym(resp_alt)
            else:
                self.result += f'^_^: Translate {resp[2]} To {self.target_language}\n{self.query_string}\n'
                self.get_result(resp)
                self.get_synonym(resp)
            if self.synonyms_en and len(resp) >= 12 and resp[11]:
                self.get_synonyms_en(resp)
            if self.definitions_en and len(resp) >= 13 and resp[12]:
                self.get_definitions(resp)
            if self.examples_en and len(resp) >= 14 and resp[13]:
                self.get_examples(resp)
            if self.result_type == 'html':
                self.result_to_html()
            else:
                self.result = self.result.replace('<b>', '').replace('</b>', '')
            return self.result.encode(self.result_code, 'ignore').decode(self.result_code)
        except requests.exceptions.ReadTimeout:
            return '╰（‵□′）╯: ReadTimeout...'
        except requests.exceptions.ProxyError:
            return '(╯‵□′)╯︵┻━┻: ProxyError...'
        except Exception as e:
            return f'Errrrrrrrrror: {e}'