blob: dbc4b5faccb019d8aaca1b687562ff1d83af210a (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
"""Event loop mixins."""
import threading
from . import events
_global_lock = threading.Lock()
class _LoopBoundedMixin:
_loop = None
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'{type(self).__name__} have already bounded to another loop')
return loop
|
