_XPointerReader
The X server's pointer, asked on a connection of its own from a worker thread, so a display server slow to answer holds that thread and never the event loop. The connection opens on first use and again after a failure.
Functions
constructor__init__() -> NoneSource Code
def __init__(self) -> None:
self._d: Any = None
self._lock = threading.Lock()Returns
Nonefuncread() -> Optional[Tuple[int, int]]Blocking. The root position, or None when the server cannot be asked.
Source Code
def read(self) -> Optional[Tuple[int, int]]:
"""Blocking. The root position, or None when the server cannot be asked."""
with self._lock:
try:
if self._d is None:
self._d = display.Display(blocking_timeout=INPUT_X_REPLY_TIMEOUT_S)
p = self._d.screen().root.query_pointer()
return p.root_x, p.root_y
except Exception as e:
logger_webrtc_input.debug(f"pointer position read failed: {e}")
self._close()
return NoneReturns
typing.Optional[typing.Tuple[int, int]]funcclose() -> NoneSource Code
def close(self) -> None:
with self._lock:
self._close()Returns
Nonefunc_close() -> NoneSource Code
def _close(self) -> None:
d, self._d = self._d, None
if d is not None:
try:
d.close()
except Exception:
passReturns
None