mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

python/qemu/console_socket.py: Add type hint annotations

Browse Source 
Finish the typing of console_socket.py with annotations and no code
changes.

Signed-off-by: John Snow <jsnow@redhat.com>
Reviewed-by: Kevin Wolf <kwolf@redhat.com>
Message-id: 20201006235817.3280413-16-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2020-10-06 19:58:12 -04:00
parent 714ac05a19
commit e35c138267
1 changed files with 9 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
python/qemu/console_socket.py
View File

@ -17,7 +17,7 @@ from collections import deque
import socket import socket
import threading import threading
import time import time
from typing import Optional from typing import Deque, Optional
class ConsoleSocket(socket.socket): class ConsoleSocket(socket.socket):
@ -30,11 +30,11 @@ class ConsoleSocket(socket.socket):
Optionally a file path can be passed in and we will also Optionally a file path can be passed in and we will also
dump the characters to this file for debugging purposes. dump the characters to this file for debugging purposes.
""" """
def __init__(self, address, file=None, drain=False): def __init__(self, address: str, file: Optional[str] = None,
self._recv_timeout_sec = 300 drain: bool = False):
self._recv_timeout_sec = 300.0 self._recv_timeout_sec = 300.0
self._sleep_time = 0.5 self._sleep_time = 0.5
self._buffer = deque() self._buffer: Deque[str] = deque()
socket.socket.__init__(self, socket.AF_UNIX, socket.SOCK_STREAM) socket.socket.__init__(self, socket.AF_UNIX, socket.SOCK_STREAM)
self.connect(address) self.connect(address)
self._logfile = None self._logfile = None
@ -45,7 +45,7 @@ class ConsoleSocket(socket.socket):
if drain: if drain:
self._drain_thread = self._thread_start() self._drain_thread = self._thread_start()
def _drain_fn(self): def _drain_fn(self) -> None:
"""Drains the socket and runs while the socket is open.""" """Drains the socket and runs while the socket is open."""
while self._open: while self._open:
try: try:
@ -56,7 +56,7 @@ class ConsoleSocket(socket.socket):
# self._open is set to False. # self._open is set to False.
time.sleep(self._sleep_time) time.sleep(self._sleep_time)
def _thread_start(self): def _thread_start(self) -> threading.Thread:
"""Kick off a thread to drain the socket.""" """Kick off a thread to drain the socket."""
# Configure socket to not block and timeout. # Configure socket to not block and timeout.
# This allows our drain thread to not block # This allows our drain thread to not block
@ -68,7 +68,7 @@ class ConsoleSocket(socket.socket):
drain_thread.start() drain_thread.start()
return drain_thread return drain_thread
def close(self): def close(self) -> None:
"""Close the base object and wait for the thread to terminate""" """Close the base object and wait for the thread to terminate"""
if self._open: if self._open:
self._open = False self._open = False
@ -80,7 +80,7 @@ class ConsoleSocket(socket.socket):
self._logfile.close() self._logfile.close()
self._logfile = None self._logfile = None
def _drain_socket(self): def _drain_socket(self) -> None:
"""process arriving characters into in memory _buffer""" """process arriving characters into in memory _buffer"""
data = socket.socket.recv(self, 1) data = socket.socket.recv(self, 1)
# latin1 is needed since there are some chars # latin1 is needed since there are some chars
@ -114,7 +114,7 @@ class ConsoleSocket(socket.socket):
# socket w/o our intervention. # socket w/o our intervention.
return chars.encode("latin1") return chars.encode("latin1")
def setblocking(self, value): def setblocking(self, value: bool) -> None:
"""When not draining we pass thru to the socket, """When not draining we pass thru to the socket,
since when draining we control socket blocking. since when draining we control socket blocking.
""" """