forked from goodboy/tractor
1
0
Fork 0

Appease `mypy`

drop_msgpack
Tyler Goodlet 2022-07-12 11:22:30 -04:00
parent bb3f35cdd0
commit 4e7ab54452
1 changed files with 31 additions and 18 deletions

View File

@ -22,10 +22,17 @@ from __future__ import annotations
import platform import platform
import struct import struct
import typing import typing
from collections.abc import AsyncGenerator, AsyncIterator from collections.abc import (
AsyncGenerator,
AsyncIterator,
)
from typing import ( from typing import (
Any, Tuple, Optional, Any,
Type, Protocol, TypeVar, runtime_checkable,
Optional,
Protocol,
Type,
TypeVar,
) )
from tricycle import BufferedReceiveStream from tricycle import BufferedReceiveStream
@ -42,7 +49,7 @@ _is_windows = platform.system() == 'Windows'
log = get_logger(__name__) log = get_logger(__name__)
def get_stream_addrs(stream: trio.SocketStream) -> Tuple: def get_stream_addrs(stream: trio.SocketStream) -> tuple:
# should both be IP sockets # should both be IP sockets
lsockname = stream.socket.getsockname() lsockname = stream.socket.getsockname()
rsockname = stream.socket.getpeername() rsockname = stream.socket.getpeername()
@ -60,6 +67,7 @@ MsgType = TypeVar("MsgType")
# - https://jcristharif.com/msgspec/usage.html#structs # - https://jcristharif.com/msgspec/usage.html#structs
@runtime_checkable
class MsgTransport(Protocol[MsgType]): class MsgTransport(Protocol[MsgType]):
stream: trio.SocketStream stream: trio.SocketStream
@ -87,15 +95,18 @@ class MsgTransport(Protocol[MsgType]):
... ...
@property @property
def laddr(self) -> Tuple[str, int]: def laddr(self) -> tuple[str, int]:
... ...
@property @property
def raddr(self) -> Tuple[str, int]: def raddr(self) -> tuple[str, int]:
... ...
class MsgpackTCPStream: # TODO: not sure why we have to inherit here, but it seems to be an
# issue with ``get_msg_transport()`` returning a ``Type[Protocol]``;
# probably should make a `mypy` issue?
class MsgpackTCPStream(MsgTransport):
''' '''
A ``trio.SocketStream`` delivering ``msgpack`` formatted data A ``trio.SocketStream`` delivering ``msgpack`` formatted data
using the ``msgspec`` codec lib. using the ``msgspec`` codec lib.
@ -190,11 +201,11 @@ class MsgpackTCPStream:
return await self.stream.send_all(size + bytes_data) return await self.stream.send_all(size + bytes_data)
@property @property
def laddr(self) -> Tuple[Any, ...]: def laddr(self) -> tuple[str, int]:
return self._laddr return self._laddr
@property @property
def raddr(self) -> Tuple[Any, ...]: def raddr(self) -> tuple[str, int]:
return self._raddr return self._raddr
async def recv(self) -> Any: async def recv(self) -> Any:
@ -223,7 +234,7 @@ class MsgpackTCPStream:
def get_msg_transport( def get_msg_transport(
key: Tuple[str, str], key: tuple[str, str],
) -> Type[MsgTransport]: ) -> Type[MsgTransport]:
@ -237,16 +248,18 @@ class Channel:
An inter-process channel for communication between (remote) actors. An inter-process channel for communication between (remote) actors.
Wraps a ``MsgStream``: transport + encoding IPC connection. Wraps a ``MsgStream``: transport + encoding IPC connection.
Currently we only support ``trio.SocketStream`` for transport Currently we only support ``trio.SocketStream`` for transport
(aka TCP). (aka TCP) and the ``msgpack`` interchange format via the ``msgspec``
codec libary.
''' '''
def __init__( def __init__(
self, self,
destaddr: Optional[Tuple[str, int]], destaddr: Optional[tuple[str, int]],
msg_transport_type_key: Tuple[str, str] = ('msgpack', 'tcp'), msg_transport_type_key: tuple[str, str] = ('msgpack', 'tcp'),
# TODO: optional reconnection support? # TODO: optional reconnection support?
# auto_reconnect: bool = False, # auto_reconnect: bool = False,
@ -266,7 +279,7 @@ class Channel:
self.msgstream: Optional[MsgTransport] = None self.msgstream: Optional[MsgTransport] = None
# set after handshake - always uid of far end # set after handshake - always uid of far end
self.uid: Optional[Tuple[str, str]] = None self.uid: Optional[tuple[str, str]] = None
self._agen = self._aiter_recv() self._agen = self._aiter_recv()
self._exc: Optional[Exception] = None # set if far end actor errors self._exc: Optional[Exception] = None # set if far end actor errors
@ -294,7 +307,7 @@ class Channel:
def set_msg_transport( def set_msg_transport(
self, self,
stream: trio.SocketStream, stream: trio.SocketStream,
type_key: Optional[Tuple[str, str]] = None, type_key: Optional[tuple[str, str]] = None,
) -> MsgTransport: ) -> MsgTransport:
type_key = type_key or self._transport_key type_key = type_key or self._transport_key
@ -309,16 +322,16 @@ class Channel:
return object.__repr__(self) return object.__repr__(self)
@property @property
def laddr(self) -> Optional[Tuple[str, int]]: def laddr(self) -> Optional[tuple[str, int]]:
return self.msgstream.laddr if self.msgstream else None return self.msgstream.laddr if self.msgstream else None
@property @property
def raddr(self) -> Optional[Tuple[str, int]]: def raddr(self) -> Optional[tuple[str, int]]:
return self.msgstream.raddr if self.msgstream else None return self.msgstream.raddr if self.msgstream else None
async def connect( async def connect(
self, self,
destaddr: Tuple[Any, ...] = None, destaddr: tuple[Any, ...] = None,
**kwargs **kwargs
) -> MsgTransport: ) -> MsgTransport: