1 """ Things the module implementing a protocol exports """
10 _ProtocolMeta, # How not to cheat here?!
20 For each class corresponding to a message, automatically create
21 two nested classes `In` and `Out` that also inherit from their
22 "nest". Class attribute `IN_KWARGS` defined in the "nest" is
23 copied to the `In` nested class under the name `KWARGS`, and
24 likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS`
25 to the nested class `Out`. In addition, methods `encode` and
26 `decode` are defined in both classes equal to `in_{en|de}code()`
27 and `out_{en|de}code()` respectively.
32 def __getattr__(self, name: str) -> Any:
35 def __setattr__(self, name: str, value: Any) -> None:
38 def in_decode(self, *args: Any) -> None:
41 def out_decode(self, *args: Any) -> None:
44 def in_encode(self, *args: Any) -> Any:
47 def out_encode(self, *args: Any) -> Any:
53 bases: Tuple[type, ...],
54 attrs: Dict[str, Any],
56 newcls = super().__new__(cls, name, bases, attrs)
57 newcls.In = super().__new__(
62 "KWARGS": newcls.IN_KWARGS,
63 "decode": newcls.in_decode,
64 "encode": newcls.in_encode,
67 newcls.Out = super().__new__(
72 "KWARGS": newcls.OUT_KWARGS,
73 "decode": newcls.out_decode,
74 "encode": newcls.out_encode,
80 # Have to do this to prevent incomprehensible error message:
81 # TypeError: metaclass conflict: the metaclass of a derived class \
82 # must be a (non-strict) subclass of the metaclasses of all its bases
83 class _MetaProto(_ProtocolMeta, MetaPkt):
87 class ProtoClass(Protocol, metaclass=_MetaProto):
88 IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
89 OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
92 def __init__(self, *args: Any, **kwargs: Any) -> None:
95 def encode(self) -> bytes:
98 def decode(self, *args: Any, **kwargs: Any) -> None:
102 def packed(self) -> bytes:
106 def __init__(self, *args: Any, **kwargs: Any) -> None:
109 def encode(self) -> bytes:
112 def decode(self, *args: Any, **kwargs: Any) -> None:
116 def packed(self) -> bytes:
122 def recv(self, segment: bytes) -> List[Union[bytes, str]]:
125 def close(self) -> bytes:
129 def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes:
132 class DecodeError(Exception):
136 def exposed_protos() -> List[Tuple[str, bool]]:
140 def probe_buffer(buffer: bytes) -> bool:
144 def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
148 def inline_response(packet: bytes) -> Optional[bytes]:
152 def is_goodbye_packet(packet: bytes) -> bool:
156 def imei_from_packet(packet: bytes) -> Optional[str]:
160 def proto_of_message(packet: bytes) -> str:
164 def proto_handled(proto: str) -> bool:
168 def class_by_prefix(prefix: str) -> Union[Type[ProtoClass], List[str]]: