import time
import math
import random
+import warnings
from inspect import isawaitable
from tornado.concurrent import (
.. versionchanged:: 5.0
This method also sets the current `asyncio` event loop.
+
+ .. deprecated:: 6.2
+ Setting and clearing the current event loop through Tornado is
+ deprecated. Use ``asyncio.set_event_loop`` instead if you need this.
"""
# The asyncio event loops override this method.
raise NotImplementedError()
.. versionchanged:: 5.0
This method also clears the current `asyncio` event loop.
+ .. deprecated:: 6.2
"""
+ warnings.warn(
+ "clear_current is deprecated",
+ DeprecationWarning,
+ stacklevel=2,
+ )
IOLoop._clear_current()
@staticmethod
super().close(all_fds=all_fds)
def make_current(self) -> None:
+ warnings.warn(
+ "make_current is deprecated; start the event loop first",
+ DeprecationWarning,
+ stacklevel=2,
+ )
if not self.is_current:
try:
self.old_asyncio = asyncio.get_event_loop()
module=r"tornado\..*",
)
super().setUp()
+ try:
+ self._prev_aio_loop = asyncio.get_event_loop()
+ except RuntimeError:
+ self._prev_aio_loop = None # type: ignore[assignment]
self.io_loop = self.get_new_ioloop()
- self.io_loop.make_current()
+ asyncio.set_event_loop(self.io_loop.asyncio_loop) # type: ignore[attr-defined]
def tearDown(self) -> None:
# Native coroutines tend to produce warnings if they're not
# Clean up Subprocess, so it can be used again with a new ioloop.
Subprocess.uninitialize()
- self.io_loop.clear_current()
+ asyncio.set_event_loop(self._prev_aio_loop)
if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):
# Try to clean up any file descriptors left open in the ioloop.
# This avoids leaks, especially when tests are run repeatedly