32 lines
803 B
Python
32 lines
803 B
Python
"""Event loop mixins."""
|
|
|
|
import threading
|
|
from . import events
|
|
|
|
_global_lock = threading.Lock()
|
|
|
|
# Used as a sentinel for loop parameter
|
|
_marker = object()
|
|
|
|
|
|
class _LoopBoundMixin:
|
|
_loop = None
|
|
|
|
def __init__(self, *, loop=_marker):
|
|
if loop is not _marker:
|
|
raise TypeError(
|
|
f'As of 3.10, the *loop* parameter was removed from '
|
|
f'{type(self).__name__}() since it is no longer necessary'
|
|
)
|
|
|
|
def _get_loop(self):
|
|
loop = events._get_running_loop()
|
|
|
|
if self._loop is None:
|
|
with _global_lock:
|
|
if self._loop is None:
|
|
self._loop = loop
|
|
if loop is not self._loop:
|
|
raise RuntimeError(f'{self!r} is bound to a different event loop')
|
|
return loop
|