GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-12 09:10:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

tools.clear_queue()

Browse Source
This commit is contained in:
Devaev Maxim 2020-10-13 17:27:36 +03:00
parent 23ff97ba95
commit bee33f2df6
3 changed files with 15 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
kvmd/plugins/hid/otg/device.py
View File

@ -31,6 +31,7 @@ from typing import Dict
from ....logging import get_logger from ....logging import get_logger
from .... import tools
from .... import aiomulti from .... import aiomulti
from .... import aioproc from .... import aioproc
@ -134,11 +135,7 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in
self.__events_queue.put_nowait(event) self.__events_queue.put_nowait(event)
def _clear_queue(self) -> None: def _clear_queue(self) -> None:
while not self.__events_queue.empty(): tools.clear_queue(self.__events_queue)
try:
self.__events_queue.get_nowait()
except queue.Empty:
break
def _ensure_write(self, report: bytes, reopen: bool=False, close: bool=False) -> bool: def _ensure_write(self, report: bytes, reopen: bool=False, close: bool=False) -> bool:
if reopen: if reopen:

7
kvmd/plugins/hid/serial.py
View File

@ -43,6 +43,7 @@ from ...logging import get_logger
from ...keyboard.mappings import KEYMAP from ...keyboard.mappings import KEYMAP
from ... import env from ... import env
from ... import tools
from ... import aiotools from ... import aiotools
from ... import aiomulti from ... import aiomulti
from ... import aioproc from ... import aioproc
@ -320,11 +321,7 @@ class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-inst
self.__queue_event(_MouseWheelEvent(delta_x, delta_y)) self.__queue_event(_MouseWheelEvent(delta_x, delta_y))
def clear_events(self) -> None: def clear_events(self) -> None:
while not self.__events_queue.empty(): tools.clear_queue(self.__events_queue)
try:
self.__events_queue.get_nowait()
except queue.Empty:
break
self.__queue_event(_ClearEvent()) self.__queue_event(_ClearEvent())
def __queue_event(self, event: _BaseEvent) -> None: def __queue_event(self, event: _BaseEvent) -> None:

11
kvmd/tools.py
View File

@ -22,6 +22,8 @@
import operator import operator
import functools import functools
import multiprocessing.queues
import queue
from typing import Tuple from typing import Tuple
from typing import List from typing import List
@ -53,3 +55,12 @@ _DictValueT = TypeVar("_DictValueT")
def sorted_kvs(dct: Dict[_DictKeyT, _DictValueT]) -> List[Tuple[_DictKeyT, _DictValueT]]: def sorted_kvs(dct: Dict[_DictKeyT, _DictValueT]) -> List[Tuple[_DictKeyT, _DictValueT]]:
return sorted(dct.items(), key=operator.itemgetter(0)) return sorted(dct.items(), key=operator.itemgetter(0))
# =====
def clear_queue(q: multiprocessing.queues.Queue) -> None: # pylint: disable=invalid-name
for _ in range(q.qsize()):
try:
q.get_nowait()
except queue.Empty:
break