diff options
Diffstat (limited to 'python/qemu/aqmp/protocol.py')
| -rw-r--r-- | python/qemu/aqmp/protocol.py | 16 |
1 files changed, 10 insertions, 6 deletions
diff --git a/python/qemu/aqmp/protocol.py b/python/qemu/aqmp/protocol.py index c4fbe35a0e..5b4f2f0d0a 100644 --- a/python/qemu/aqmp/protocol.py +++ b/python/qemu/aqmp/protocol.py @@ -46,6 +46,10 @@ T = TypeVar('T') _U = TypeVar('_U') _TaskFN = Callable[[], Awaitable[None]] # aka ``async def func() -> None`` +InternetAddrT = Tuple[str, int] +UnixAddrT = str +SocketAddrT = Union[UnixAddrT, InternetAddrT] + class Runstate(Enum): """Protocol session runstate.""" @@ -257,7 +261,7 @@ class AsyncProtocol(Generic[T]): @upper_half @require(Runstate.IDLE) - async def accept(self, address: Union[str, Tuple[str, int]], + async def accept(self, address: SocketAddrT, ssl: Optional[SSLContext] = None) -> None: """ Accept a connection and begin processing message queues. @@ -275,7 +279,7 @@ class AsyncProtocol(Generic[T]): @upper_half @require(Runstate.IDLE) - async def connect(self, address: Union[str, Tuple[str, int]], + async def connect(self, address: SocketAddrT, ssl: Optional[SSLContext] = None) -> None: """ Connect to the server and begin processing message queues. @@ -337,7 +341,7 @@ class AsyncProtocol(Generic[T]): @upper_half async def _new_session(self, - address: Union[str, Tuple[str, int]], + address: SocketAddrT, ssl: Optional[SSLContext] = None, accept: bool = False) -> None: """ @@ -397,7 +401,7 @@ class AsyncProtocol(Generic[T]): @upper_half async def _establish_connection( self, - address: Union[str, Tuple[str, int]], + address: SocketAddrT, ssl: Optional[SSLContext] = None, accept: bool = False ) -> None: @@ -424,7 +428,7 @@ class AsyncProtocol(Generic[T]): await self._do_connect(address, ssl) @upper_half - async def _do_accept(self, address: Union[str, Tuple[str, int]], + async def _do_accept(self, address: SocketAddrT, ssl: Optional[SSLContext] = None) -> None: """ Acting as the transport server, accept a single connection. @@ -482,7 +486,7 @@ class AsyncProtocol(Generic[T]): self.logger.debug("Connection accepted.") @upper_half - async def _do_connect(self, address: Union[str, Tuple[str, int]], + async def _do_connect(self, address: SocketAddrT, ssl: Optional[SSLContext] = None) -> None: """ Acting as the transport client, initiate a connection to a server. |