Run a coroutine in the asyncio event loop from a separate thread. :param coro: The coroutine to be run. :return: The result of the coroutine.
(self, coro)
| 923 | self.loop.run_forever() |
| 924 | |
| 925 | def run_coro(self, coro) -> Any: |
| 926 | """ |
| 927 | Run a coroutine in the asyncio event loop from a separate thread. |
| 928 | |
| 929 | :param coro: The coroutine to be run. |
| 930 | |
| 931 | :return: The result of the coroutine. |
| 932 | """ |
| 933 | return asyncio.run_coroutine_threadsafe(coro, loop=self.loop).result() |
| 934 | |
| 935 | def stop(self) -> None: |
| 936 | """ |
no outgoing calls
no test coverage detected