efro.message
Functionality for sending and responding to messages. Supports static typing for message types and possible return types.
1# Released under the MIT License. See LICENSE for details. 2# 3"""Functionality for sending and responding to messages. 4Supports static typing for message types and possible return types. 5""" 6 7from __future__ import annotations 8 9from efro.util import set_canonical_module_names 10from efro.message._protocol import MessageProtocol 11from efro.message._sender import MessageSender, BoundMessageSender 12from efro.message._receiver import MessageReceiver, BoundMessageReceiver 13from efro.message._module import create_sender_module, create_receiver_module 14from efro.message._message import ( 15 Message, 16 Response, 17 SysResponse, 18 EmptySysResponse, 19 ErrorSysResponse, 20 StringResponse, 21 BoolResponse, 22 UnregisteredMessageIDError, 23) 24 25__all__ = [ 26 'Message', 27 'Response', 28 'SysResponse', 29 'EmptySysResponse', 30 'ErrorSysResponse', 31 'StringResponse', 32 'BoolResponse', 33 'MessageProtocol', 34 'MessageSender', 35 'BoundMessageSender', 36 'MessageReceiver', 37 'BoundMessageReceiver', 38 'create_sender_module', 39 'create_receiver_module', 40 'UnregisteredMessageIDError', 41] 42 43# Have these things present themselves cleanly as 'thismodule.SomeClass' 44# instead of 'thismodule._internalmodule.SomeClass' 45set_canonical_module_names(globals())
class
Message:
24class Message: 25 """Base class for messages.""" 26 27 @classmethod 28 def get_response_types(cls) -> list[type[Response] | None]: 29 """Return all Response types this Message can return when sent. 30 31 The default implementation specifies a None return type. 32 """ 33 return [None]
Base class for messages.
27 @classmethod 28 def get_response_types(cls) -> list[type[Response] | None]: 29 """Return all Response types this Message can return when sent. 30 31 The default implementation specifies a None return type. 32 """ 33 return [None]
Return all Response types this Message can return when sent.
The default implementation specifies a None return type.
class
Response:
Base class for responses to messages.
class
SysResponse:
40class SysResponse: 41 """Base class for system-responses to messages. 42 43 These are only sent/handled by the messaging system itself; 44 users of the api never see them. 45 """ 46 47 def set_local_exception(self, exc: Exception) -> None: 48 """Attach a local exception to facilitate better logging/handling. 49 50 Be aware that this data does not get serialized and only 51 exists on the local object. 52 """ 53 setattr(self, '_sr_local_exception', exc) 54 55 def get_local_exception(self) -> Exception | None: 56 """Fetch a local attached exception.""" 57 value = getattr(self, '_sr_local_exception', None) 58 assert isinstance(value, Exception | None) 59 return value
Base class for system-responses to messages.
These are only sent/handled by the messaging system itself; users of the api never see them.
def
set_local_exception(self, exc: Exception) -> None:
47 def set_local_exception(self, exc: Exception) -> None: 48 """Attach a local exception to facilitate better logging/handling. 49 50 Be aware that this data does not get serialized and only 51 exists on the local object. 52 """ 53 setattr(self, '_sr_local_exception', exc)
Attach a local exception to facilitate better logging/handling.
Be aware that this data does not get serialized and only exists on the local object.
86@ioprepped 87@dataclass 88class EmptySysResponse(SysResponse): 89 """The response equivalent of None."""
The response equivalent of None.
Inherited Members
65@ioprepped 66@dataclass 67class ErrorSysResponse(SysResponse): 68 """SysResponse saying some error has occurred for the send. 69 70 This generally results in an Exception being raised for the caller. 71 """ 72 73 class ErrorType(Enum): 74 """Type of error that occurred while sending a message.""" 75 76 REMOTE = 0 77 REMOTE_CLEAN = 1 78 LOCAL = 2 79 COMMUNICATION = 3 80 REMOTE_COMMUNICATION = 4 81 82 error_message: Annotated[str, IOAttrs('m')] 83 error_type: Annotated[ErrorType, IOAttrs('e')] = ErrorType.REMOTE
SysResponse saying some error has occurred for the send.
This generally results in an Exception being raised for the caller.
ErrorSysResponse( error_message: Annotated[str, <efro.dataclassio.IOAttrs object>], error_type: Annotated[ErrorSysResponse.ErrorType, <efro.dataclassio.IOAttrs object>] = <ErrorType.REMOTE: 0>)
error_message: Annotated[str, <efro.dataclassio.IOAttrs object at 0x10fa46120>]
error_type: Annotated[ErrorSysResponse.ErrorType, <efro.dataclassio.IOAttrs object at 0x10fa47ad0>] =
<ErrorType.REMOTE: 0>
Inherited Members
class
ErrorSysResponse.ErrorType(enum.Enum):
73 class ErrorType(Enum): 74 """Type of error that occurred while sending a message.""" 75 76 REMOTE = 0 77 REMOTE_CLEAN = 1 78 LOCAL = 2 79 COMMUNICATION = 3 80 REMOTE_COMMUNICATION = 4
Type of error that occurred while sending a message.
Inherited Members
- enum.Enum
- name
- value
104@ioprepped 105@dataclass 106class StringResponse(Response): 107 """A simple string value response.""" 108 109 value: Annotated[str, IOAttrs('v')]
A simple string value response.
StringResponse(value: Annotated[str, <efro.dataclassio.IOAttrs object>])
value: Annotated[str, <efro.dataclassio.IOAttrs object at 0x11042a750>]
96@ioprepped 97@dataclass 98class BoolResponse(Response): 99 """A simple bool value response.""" 100 101 value: Annotated[bool, IOAttrs('v')]
A simple bool value response.
BoolResponse(value: Annotated[bool, <efro.dataclassio.IOAttrs object>])
value: Annotated[bool, <efro.dataclassio.IOAttrs object at 0x1110bc680>]
class
MessageProtocol:
33class MessageProtocol: 34 """Wrangles a set of message types, formats, and response types. 35 Both endpoints must be using a compatible Protocol for communication 36 to succeed. To maintain Protocol compatibility between revisions, 37 all message types must retain the same id, message attr storage 38 names must not change, newly added attrs must have default values, 39 etc. 40 """ 41 42 def __init__( 43 self, 44 message_types: dict[int, type[Message]], 45 response_types: dict[int, type[Response]], 46 forward_communication_errors: bool = False, 47 forward_clean_errors: bool = False, 48 remote_errors_include_stack_traces: bool = False, 49 log_errors_on_receiver: bool = True, 50 ) -> None: 51 """Create a protocol with a given configuration. 52 53 If 'forward_communication_errors' is True, 54 efro.error.CommunicationErrors raised on the receiver end will 55 result in a matching error raised back on the sender. This can 56 be useful if the receiver will be in some way forwarding 57 messages along and the sender doesn't need to know where 58 communication breakdowns occurred; only that they did. 59 60 If 'forward_clean_errors' is True, efro.error.CleanError 61 exceptions raised on the receiver end will result in a matching 62 CleanError raised back on the sender. 63 64 When an exception is not covered by the optional forwarding 65 mechanisms above, it will come across as efro.error.RemoteError 66 and the exception will be logged on the receiver end - at least 67 by default (see details below). 68 69 If 'remote_errors_include_stack_traces' is True, stringified 70 stack traces will be returned with efro.error.RemoteError 71 exceptions. This is useful for debugging but should only be 72 enabled in cases where the sender is trusted to see internal 73 details of the receiver. 74 75 By default, when a message-handling exception will result in an 76 efro.error.RemoteError being returned to the sender, the 77 exception will be logged on the receiver. This is because the 78 goal is usually to avoid returning opaque RemoteErrors and to 79 instead return something meaningful as part of the expected 80 response type (even if that value itself represents a logical 81 error state). If 'log_errors_on_receiver' is False, however, such 82 exceptions will *not* be logged on the receiver. This can be 83 useful in combination with 'remote_errors_include_stack_traces' 84 and 'forward_clean_errors' in situations where all error 85 logging/management will be happening on the sender end. Be 86 aware, however, that in that case it may be possible for 87 communication errors to prevent such error messages from 88 ever being seen. 89 """ 90 # pylint: disable=too-many-locals 91 self.message_types_by_id: dict[int, type[Message]] = {} 92 self.message_ids_by_type: dict[type[Message], int] = {} 93 self.response_types_by_id: dict[ 94 int, type[Response] | type[SysResponse] 95 ] = {} 96 self.response_ids_by_type: dict[ 97 type[Response] | type[SysResponse], int 98 ] = {} 99 for m_id, m_type in message_types.items(): 100 # Make sure only valid message types were passed and each 101 # id was assigned only once. 102 assert isinstance(m_id, int) 103 assert m_id >= 0 104 assert is_ioprepped_dataclass(m_type) and issubclass( 105 m_type, Message 106 ) 107 assert self.message_types_by_id.get(m_id) is None 108 self.message_types_by_id[m_id] = m_type 109 self.message_ids_by_type[m_type] = m_id 110 111 for r_id, r_type in response_types.items(): 112 assert isinstance(r_id, int) 113 assert r_id >= 0 114 assert is_ioprepped_dataclass(r_type) and issubclass( 115 r_type, Response 116 ) 117 assert self.response_types_by_id.get(r_id) is None 118 self.response_types_by_id[r_id] = r_type 119 self.response_ids_by_type[r_type] = r_id 120 121 # Register our SysResponse types. These use negative 122 # IDs so as to never overlap with user Response types. 123 def _reg_sys(reg_tp: type[SysResponse], reg_id: int) -> None: 124 assert self.response_types_by_id.get(reg_id) is None 125 self.response_types_by_id[reg_id] = reg_tp 126 self.response_ids_by_type[reg_tp] = reg_id 127 128 _reg_sys(ErrorSysResponse, -1) 129 _reg_sys(EmptySysResponse, -2) 130 131 # Some extra-thorough validation in debug mode. 132 if __debug__: 133 # Make sure all Message types' return types are valid 134 # and have been assigned an ID as well. 135 all_response_types: set[type[Response] | None] = set() 136 for m_id, m_type in message_types.items(): 137 m_rtypes = m_type.get_response_types() 138 139 assert isinstance(m_rtypes, list) 140 assert ( 141 m_rtypes 142 ), f'Message type {m_type} specifies no return types.' 143 assert len(set(m_rtypes)) == len(m_rtypes) # check dups 144 for m_rtype in m_rtypes: 145 all_response_types.add(m_rtype) 146 for cls in all_response_types: 147 if cls is None: 148 continue 149 assert is_ioprepped_dataclass(cls) 150 assert issubclass(cls, Response) 151 if cls not in self.response_ids_by_type: 152 raise ValueError( 153 f'Possible response type {cls} needs to be included' 154 f' in response_types for this protocol.' 155 ) 156 157 # Make sure all registered types have unique base names. 158 # We can take advantage of this to generate cleaner looking 159 # protocol modules. Can revisit if this is ever a problem. 160 mtypenames = set(tp.__name__ for tp in self.message_ids_by_type) 161 if len(mtypenames) != len(message_types): 162 raise ValueError( 163 'message_types contains duplicate __name__s;' 164 ' all types are required to have unique names.' 165 ) 166 167 self.forward_clean_errors = forward_clean_errors 168 self.forward_communication_errors = forward_communication_errors 169 self.remote_errors_include_stack_traces = ( 170 remote_errors_include_stack_traces 171 ) 172 self.log_errors_on_receiver = log_errors_on_receiver 173 174 @staticmethod 175 def encode_dict(obj: dict) -> str: 176 """Json-encode a provided dict.""" 177 return json.dumps(obj, separators=(',', ':')) 178 179 def message_to_dict(self, message: Message) -> dict: 180 """Encode a message to a json ready dict.""" 181 return self._to_dict(message, self.message_ids_by_type, 'message') 182 183 def response_to_dict(self, response: Response | SysResponse) -> dict: 184 """Encode a response to a json ready dict.""" 185 return self._to_dict(response, self.response_ids_by_type, 'response') 186 187 def error_to_response(self, exc: Exception) -> tuple[SysResponse, bool]: 188 """Translate an Exception to a SysResponse. 189 190 Also returns whether the error should be logged if this happened 191 within handle_raw_message(). 192 """ 193 194 # If anything goes wrong, return a ErrorSysResponse instead. 195 # (either CLEAN or generic REMOTE) 196 if self.forward_clean_errors and isinstance(exc, CleanError): 197 return ( 198 ErrorSysResponse( 199 error_message=str(exc), 200 error_type=ErrorSysResponse.ErrorType.REMOTE_CLEAN, 201 ), 202 False, 203 ) 204 if self.forward_communication_errors and isinstance( 205 exc, CommunicationError 206 ): 207 return ( 208 ErrorSysResponse( 209 error_message=str(exc), 210 error_type=ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION, 211 ), 212 False, 213 ) 214 return ( 215 ErrorSysResponse( 216 error_message=( 217 # Note: need to format exception ourself here; it 218 # might not be current so we can't use 219 # traceback.format_exc(). 220 ''.join( 221 traceback.format_exception( 222 type(exc), exc, exc.__traceback__ 223 ) 224 ) 225 if self.remote_errors_include_stack_traces 226 else 'An internal error has occurred.' 227 ), 228 error_type=ErrorSysResponse.ErrorType.REMOTE, 229 ), 230 self.log_errors_on_receiver, 231 ) 232 233 def _to_dict( 234 self, message: Any, ids_by_type: dict[type, int], opname: str 235 ) -> dict: 236 """Encode a message to a json string for transport.""" 237 238 m_id: int | None = ids_by_type.get(type(message)) 239 if m_id is None: 240 raise TypeError( 241 f'{opname} type is not registered in protocol:' 242 f' {type(message)}' 243 ) 244 out = {'t': m_id, 'm': dataclass_to_dict(message)} 245 return out 246 247 @staticmethod 248 def decode_dict(data: str) -> dict: 249 """Decode data to a dict.""" 250 out = json.loads(data) 251 assert isinstance(out, dict) 252 return out 253 254 def message_from_dict(self, data: dict) -> Message: 255 """Decode a message from a json string.""" 256 out = self._from_dict(data, self.message_types_by_id, 'message') 257 assert isinstance(out, Message) 258 return out 259 260 def response_from_dict(self, data: dict) -> Response | SysResponse: 261 """Decode a response from a json string.""" 262 out = self._from_dict(data, self.response_types_by_id, 'response') 263 assert isinstance(out, Response | SysResponse) 264 return out 265 266 # Weeeird; we get mypy errors returning dict[int, type] but 267 # dict[int, typing.Type] or dict[int, type[Any]] works.. 268 def _from_dict( 269 self, data: dict, types_by_id: dict[int, type[Any]], opname: str 270 ) -> Any: 271 """Decode a message from a json string.""" 272 msgdict: dict | None 273 274 m_id = data.get('t') 275 # Allow omitting 'm' dict if its empty. 276 msgdict = data.get('m', {}) 277 278 assert isinstance(m_id, int) 279 assert isinstance(msgdict, dict) 280 281 # Decode this particular type. 282 msgtype = types_by_id.get(m_id) 283 if msgtype is None: 284 raise UnregisteredMessageIDError( 285 f'Got unregistered {opname} id of {m_id}.' 286 ) 287 return dataclass_from_dict(msgtype, msgdict) 288 289 def _get_module_header( 290 self, 291 part: Literal['sender', 'receiver'], 292 extra_import_code: str | None, 293 enable_async_sends: bool, 294 ) -> str: 295 """Return common parts of generated modules.""" 296 # pylint: disable=too-many-locals 297 # pylint: disable=too-many-branches 298 # pylint: disable=too-many-statements 299 import textwrap 300 301 tpimports: dict[str, list[str]] = {} 302 imports: dict[str, list[str]] = {} 303 304 single_message_type = len(self.message_ids_by_type) == 1 305 306 msgtypes = list(self.message_ids_by_type) 307 if part == 'sender': 308 msgtypes.append(Message) 309 for msgtype in msgtypes: 310 tpimports.setdefault(msgtype.__module__, []).append( 311 msgtype.__name__ 312 ) 313 rsptypes = list(self.response_ids_by_type) 314 if part == 'sender': 315 rsptypes.append(Response) 316 for rsp_tp in rsptypes: 317 # Skip these as they don't actually show up in code. 318 if rsp_tp is EmptySysResponse or rsp_tp is ErrorSysResponse: 319 continue 320 if ( 321 single_message_type 322 and part == 'sender' 323 and rsp_tp is not Response 324 ): 325 # We need to cast to the single supported response type 326 # in this case so need response types at runtime. 327 imports.setdefault(rsp_tp.__module__, []).append( 328 rsp_tp.__name__ 329 ) 330 else: 331 tpimports.setdefault(rsp_tp.__module__, []).append( 332 rsp_tp.__name__ 333 ) 334 335 import_lines = '' 336 tpimport_lines = '' 337 338 for module, names in sorted(imports.items()): 339 jnames = ', '.join(names) 340 line = f'from {module} import {jnames}' 341 if len(line) > 80: 342 # Recreate in a wrapping-friendly form. 343 line = f'from {module} import ({jnames})' 344 import_lines += f'{line}\n' 345 for module, names in sorted(tpimports.items()): 346 jnames = ', '.join(names) 347 line = f'from {module} import {jnames}' 348 if len(line) > 75: # Account for indent 349 # Recreate in a wrapping-friendly form. 350 line = f'from {module} import ({jnames})' 351 tpimport_lines += f'{line}\n' 352 353 if part == 'sender': 354 import_lines += ( 355 'from efro.message import MessageSender, BoundMessageSender\n' 356 ) 357 tpimport_typing_extras = '' 358 else: 359 if single_message_type: 360 import_lines += ( 361 'from efro.message import (MessageReceiver,' 362 ' BoundMessageReceiver, Message, Response)\n' 363 ) 364 else: 365 import_lines += ( 366 'from efro.message import MessageReceiver,' 367 ' BoundMessageReceiver\n' 368 ) 369 tpimport_typing_extras = ', Awaitable' 370 371 if extra_import_code is not None: 372 import_lines += extra_import_code 373 374 ovld = ', overload' if not single_message_type else '' 375 ovld2 = ( 376 ', cast, Awaitable' 377 if (single_message_type and part == 'sender' and enable_async_sends) 378 else '' 379 ) 380 tpimport_lines = textwrap.indent(tpimport_lines, ' ') 381 382 baseimps = ['Any'] 383 if part == 'receiver': 384 baseimps.append('Callable') 385 if part == 'sender' and enable_async_sends: 386 baseimps.append('Awaitable') 387 baseimps_s = ', '.join(baseimps) 388 out = ( 389 '# Released under the MIT License. See LICENSE for details.\n' 390 f'#\n' 391 f'"""Auto-generated {part} module. Do not edit by hand."""\n' 392 f'\n' 393 f'from __future__ import annotations\n' 394 f'\n' 395 f'from typing import TYPE_CHECKING{ovld}{ovld2}\n' 396 f'\n' 397 f'{import_lines}' 398 f'\n' 399 f'if TYPE_CHECKING:\n' 400 f' from typing import {baseimps_s}' 401 f'{tpimport_typing_extras}\n' 402 f'{tpimport_lines}' 403 f'\n' 404 f'\n' 405 ) 406 return out 407 408 def do_create_sender_module( 409 self, 410 basename: str, 411 protocol_create_code: str, 412 enable_sync_sends: bool, 413 enable_async_sends: bool, 414 private: bool = False, 415 protocol_module_level_import_code: str | None = None, 416 ) -> str: 417 """Used by create_sender_module(); do not call directly.""" 418 # pylint: disable=too-many-locals 419 # pylint: disable=too-many-branches 420 import textwrap 421 422 msgtypes = list(self.message_ids_by_type.keys()) 423 424 ppre = '_' if private else '' 425 out = self._get_module_header( 426 'sender', 427 extra_import_code=protocol_module_level_import_code, 428 enable_async_sends=enable_async_sends, 429 ) 430 ccind = textwrap.indent(protocol_create_code, ' ') 431 out += ( 432 f'class {ppre}{basename}(MessageSender):\n' 433 f' """Protocol-specific sender."""\n' 434 f'\n' 435 f' def __init__(self) -> None:\n' 436 f'{ccind}\n' 437 f' super().__init__(protocol)\n' 438 f'\n' 439 f' def __get__(\n' 440 f' self, obj: Any, type_in: Any = None\n' 441 f' ) -> {ppre}Bound{basename}:\n' 442 f' return {ppre}Bound{basename}(obj, self)\n' 443 f'\n' 444 f'\n' 445 f'class {ppre}Bound{basename}(BoundMessageSender):\n' 446 f' """Protocol-specific bound sender."""\n' 447 ) 448 449 def _filt_tp_name(rtype: type[Response] | None) -> str: 450 return 'None' if rtype is None else rtype.__name__ 451 452 # Define handler() overloads for all registered message types. 453 if msgtypes: 454 for async_pass in False, True: 455 if async_pass and not enable_async_sends: 456 continue 457 if not async_pass and not enable_sync_sends: 458 continue 459 pfx = 'async ' if async_pass else '' 460 sfx = '_async' if async_pass else '' 461 # awt = 'await ' if async_pass else '' 462 awt = '' 463 how = 'asynchronously' if async_pass else 'synchronously' 464 465 if len(msgtypes) == 1: 466 # Special case: with a single message types we don't 467 # use overloads. 468 msgtype = msgtypes[0] 469 msgtypevar = msgtype.__name__ 470 rtypes = msgtype.get_response_types() 471 if len(rtypes) > 1: 472 rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes) 473 else: 474 rtypevar = _filt_tp_name(rtypes[0]) 475 if async_pass: 476 rtypevar = f'Awaitable[{rtypevar}]' 477 out += ( 478 f'\n' 479 f' def send{sfx}(self,' 480 f' message: {msgtypevar})' 481 f' -> {rtypevar}:\n' 482 f' """Send a message {how}."""\n' 483 f' out = {awt}self._sender.' 484 f'send{sfx}(self._obj, message)\n' 485 ) 486 if not async_pass: 487 out += ( 488 f' assert isinstance(out, {rtypevar})\n' 489 ' return out\n' 490 ) 491 else: 492 out += f' return cast({rtypevar}, out)\n' 493 494 else: 495 for msgtype in msgtypes: 496 msgtypevar = msgtype.__name__ 497 rtypes = msgtype.get_response_types() 498 if len(rtypes) > 1: 499 rtypevar = ' | '.join( 500 _filt_tp_name(t) for t in rtypes 501 ) 502 else: 503 rtypevar = _filt_tp_name(rtypes[0]) 504 out += ( 505 f'\n' 506 f' @overload\n' 507 f' {pfx}def send{sfx}(self,' 508 f' message: {msgtypevar})' 509 f' -> {rtypevar}: ...\n' 510 ) 511 rtypevar = 'Response | None' 512 if async_pass: 513 rtypevar = f'Awaitable[{rtypevar}]' 514 out += ( 515 f'\n' 516 f' def send{sfx}(self, message: Message)' 517 f' -> {rtypevar}:\n' 518 f' """Send a message {how}."""\n' 519 f' return {awt}self._sender.' 520 f'send{sfx}(self._obj, message)\n' 521 ) 522 523 return out 524 525 def do_create_receiver_module( 526 self, 527 basename: str, 528 protocol_create_code: str, 529 is_async: bool, 530 private: bool = False, 531 protocol_module_level_import_code: str | None = None, 532 ) -> str: 533 """Used by create_receiver_module(); do not call directly.""" 534 # pylint: disable=too-many-locals 535 import textwrap 536 537 desc = 'asynchronous' if is_async else 'synchronous' 538 ppre = '_' if private else '' 539 msgtypes = list(self.message_ids_by_type.keys()) 540 out = self._get_module_header( 541 'receiver', 542 extra_import_code=protocol_module_level_import_code, 543 enable_async_sends=False, 544 ) 545 ccind = textwrap.indent(protocol_create_code, ' ') 546 out += ( 547 f'class {ppre}{basename}(MessageReceiver):\n' 548 f' """Protocol-specific {desc} receiver."""\n' 549 f'\n' 550 f' is_async = {is_async}\n' 551 f'\n' 552 f' def __init__(self) -> None:\n' 553 f'{ccind}\n' 554 f' super().__init__(protocol)\n' 555 f'\n' 556 f' def __get__(\n' 557 f' self,\n' 558 f' obj: Any,\n' 559 f' type_in: Any = None,\n' 560 f' ) -> {ppre}Bound{basename}:\n' 561 f' return {ppre}Bound{basename}(' 562 f'obj, self)\n' 563 ) 564 565 # Define handler() overloads for all registered message types. 566 567 def _filt_tp_name(rtype: type[Response] | None) -> str: 568 return 'None' if rtype is None else rtype.__name__ 569 570 if msgtypes: 571 cbgn = 'Awaitable[' if is_async else '' 572 cend = ']' if is_async else '' 573 if len(msgtypes) == 1: 574 # Special case: when we have a single message type we don't 575 # use overloads. 576 msgtype = msgtypes[0] 577 msgtypevar = msgtype.__name__ 578 rtypes = msgtype.get_response_types() 579 if len(rtypes) > 1: 580 rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes) 581 else: 582 rtypevar = _filt_tp_name(rtypes[0]) 583 rtypevar = f'{cbgn}{rtypevar}{cend}' 584 out += ( 585 f'\n' 586 f' def handler(\n' 587 f' self,\n' 588 f' call: Callable[[Any, {msgtypevar}], ' 589 f'{rtypevar}],\n' 590 f' )' 591 f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n' 592 f' """Decorator to register message handlers."""\n' 593 f' from typing import cast, Callable, Any\n' 594 f'\n' 595 f' self.register_handler(cast(Callable' 596 f'[[Any, Message], Response], call))\n' 597 f' return call\n' 598 ) 599 else: 600 for msgtype in msgtypes: 601 msgtypevar = msgtype.__name__ 602 rtypes = msgtype.get_response_types() 603 if len(rtypes) > 1: 604 rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes) 605 else: 606 rtypevar = _filt_tp_name(rtypes[0]) 607 rtypevar = f'{cbgn}{rtypevar}{cend}' 608 out += ( 609 f'\n' 610 f' @overload\n' 611 f' def handler(\n' 612 f' self,\n' 613 f' call: Callable[[Any, {msgtypevar}], ' 614 f'{rtypevar}],\n' 615 f' )' 616 f' -> Callable[[Any, {msgtypevar}], {rtypevar}]: ...\n' 617 ) 618 out += ( 619 '\n' 620 ' def handler(self, call: Callable) -> Callable:\n' 621 ' """Decorator to register message handlers."""\n' 622 ' self.register_handler(call)\n' 623 ' return call\n' 624 ) 625 626 out += ( 627 f'\n' 628 f'\n' 629 f'class {ppre}Bound{basename}(BoundMessageReceiver):\n' 630 f' """Protocol-specific bound receiver."""\n' 631 ) 632 if is_async: 633 out += ( 634 '\n' 635 ' def handle_raw_message(\n' 636 ' self, message: str, raise_unregistered: bool = False\n' 637 ' ) -> Awaitable[str]:\n' 638 ' """Asynchronously handle a raw incoming message."""\n' 639 ' return self._receiver.' 640 'handle_raw_message_async(\n' 641 ' self._obj, message, raise_unregistered\n' 642 ' )\n' 643 ) 644 645 else: 646 out += ( 647 '\n' 648 ' def handle_raw_message(\n' 649 ' self, message: str, raise_unregistered: bool = False\n' 650 ' ) -> str:\n' 651 ' """Synchronously handle a raw incoming message."""\n' 652 ' return self._receiver.handle_raw_message(\n' 653 ' self._obj, message, raise_unregistered\n' 654 ' )\n' 655 ) 656 657 return out
Wrangles a set of message types, formats, and response types. Both endpoints must be using a compatible Protocol for communication to succeed. To maintain Protocol compatibility between revisions, all message types must retain the same id, message attr storage names must not change, newly added attrs must have default values, etc.
MessageProtocol( message_types: dict[int, type[Message]], response_types: dict[int, type[Response]], forward_communication_errors: bool = False, forward_clean_errors: bool = False, remote_errors_include_stack_traces: bool = False, log_errors_on_receiver: bool = True)
42 def __init__( 43 self, 44 message_types: dict[int, type[Message]], 45 response_types: dict[int, type[Response]], 46 forward_communication_errors: bool = False, 47 forward_clean_errors: bool = False, 48 remote_errors_include_stack_traces: bool = False, 49 log_errors_on_receiver: bool = True, 50 ) -> None: 51 """Create a protocol with a given configuration. 52 53 If 'forward_communication_errors' is True, 54 efro.error.CommunicationErrors raised on the receiver end will 55 result in a matching error raised back on the sender. This can 56 be useful if the receiver will be in some way forwarding 57 messages along and the sender doesn't need to know where 58 communication breakdowns occurred; only that they did. 59 60 If 'forward_clean_errors' is True, efro.error.CleanError 61 exceptions raised on the receiver end will result in a matching 62 CleanError raised back on the sender. 63 64 When an exception is not covered by the optional forwarding 65 mechanisms above, it will come across as efro.error.RemoteError 66 and the exception will be logged on the receiver end - at least 67 by default (see details below). 68 69 If 'remote_errors_include_stack_traces' is True, stringified 70 stack traces will be returned with efro.error.RemoteError 71 exceptions. This is useful for debugging but should only be 72 enabled in cases where the sender is trusted to see internal 73 details of the receiver. 74 75 By default, when a message-handling exception will result in an 76 efro.error.RemoteError being returned to the sender, the 77 exception will be logged on the receiver. This is because the 78 goal is usually to avoid returning opaque RemoteErrors and to 79 instead return something meaningful as part of the expected 80 response type (even if that value itself represents a logical 81 error state). If 'log_errors_on_receiver' is False, however, such 82 exceptions will *not* be logged on the receiver. This can be 83 useful in combination with 'remote_errors_include_stack_traces' 84 and 'forward_clean_errors' in situations where all error 85 logging/management will be happening on the sender end. Be 86 aware, however, that in that case it may be possible for 87 communication errors to prevent such error messages from 88 ever being seen. 89 """ 90 # pylint: disable=too-many-locals 91 self.message_types_by_id: dict[int, type[Message]] = {} 92 self.message_ids_by_type: dict[type[Message], int] = {} 93 self.response_types_by_id: dict[ 94 int, type[Response] | type[SysResponse] 95 ] = {} 96 self.response_ids_by_type: dict[ 97 type[Response] | type[SysResponse], int 98 ] = {} 99 for m_id, m_type in message_types.items(): 100 # Make sure only valid message types were passed and each 101 # id was assigned only once. 102 assert isinstance(m_id, int) 103 assert m_id >= 0 104 assert is_ioprepped_dataclass(m_type) and issubclass( 105 m_type, Message 106 ) 107 assert self.message_types_by_id.get(m_id) is None 108 self.message_types_by_id[m_id] = m_type 109 self.message_ids_by_type[m_type] = m_id 110 111 for r_id, r_type in response_types.items(): 112 assert isinstance(r_id, int) 113 assert r_id >= 0 114 assert is_ioprepped_dataclass(r_type) and issubclass( 115 r_type, Response 116 ) 117 assert self.response_types_by_id.get(r_id) is None 118 self.response_types_by_id[r_id] = r_type 119 self.response_ids_by_type[r_type] = r_id 120 121 # Register our SysResponse types. These use negative 122 # IDs so as to never overlap with user Response types. 123 def _reg_sys(reg_tp: type[SysResponse], reg_id: int) -> None: 124 assert self.response_types_by_id.get(reg_id) is None 125 self.response_types_by_id[reg_id] = reg_tp 126 self.response_ids_by_type[reg_tp] = reg_id 127 128 _reg_sys(ErrorSysResponse, -1) 129 _reg_sys(EmptySysResponse, -2) 130 131 # Some extra-thorough validation in debug mode. 132 if __debug__: 133 # Make sure all Message types' return types are valid 134 # and have been assigned an ID as well. 135 all_response_types: set[type[Response] | None] = set() 136 for m_id, m_type in message_types.items(): 137 m_rtypes = m_type.get_response_types() 138 139 assert isinstance(m_rtypes, list) 140 assert ( 141 m_rtypes 142 ), f'Message type {m_type} specifies no return types.' 143 assert len(set(m_rtypes)) == len(m_rtypes) # check dups 144 for m_rtype in m_rtypes: 145 all_response_types.add(m_rtype) 146 for cls in all_response_types: 147 if cls is None: 148 continue 149 assert is_ioprepped_dataclass(cls) 150 assert issubclass(cls, Response) 151 if cls not in self.response_ids_by_type: 152 raise ValueError( 153 f'Possible response type {cls} needs to be included' 154 f' in response_types for this protocol.' 155 ) 156 157 # Make sure all registered types have unique base names. 158 # We can take advantage of this to generate cleaner looking 159 # protocol modules. Can revisit if this is ever a problem. 160 mtypenames = set(tp.__name__ for tp in self.message_ids_by_type) 161 if len(mtypenames) != len(message_types): 162 raise ValueError( 163 'message_types contains duplicate __name__s;' 164 ' all types are required to have unique names.' 165 ) 166 167 self.forward_clean_errors = forward_clean_errors 168 self.forward_communication_errors = forward_communication_errors 169 self.remote_errors_include_stack_traces = ( 170 remote_errors_include_stack_traces 171 ) 172 self.log_errors_on_receiver = log_errors_on_receiver
Create a protocol with a given configuration.
If 'forward_communication_errors' is True, efro.error.CommunicationErrors raised on the receiver end will result in a matching error raised back on the sender. This can be useful if the receiver will be in some way forwarding messages along and the sender doesn't need to know where communication breakdowns occurred; only that they did.
If 'forward_clean_errors' is True, efro.error.CleanError exceptions raised on the receiver end will result in a matching CleanError raised back on the sender.
When an exception is not covered by the optional forwarding mechanisms above, it will come across as efro.error.RemoteError and the exception will be logged on the receiver end - at least by default (see details below).
If 'remote_errors_include_stack_traces' is True, stringified stack traces will be returned with efro.error.RemoteError exceptions. This is useful for debugging but should only be enabled in cases where the sender is trusted to see internal details of the receiver.
By default, when a message-handling exception will result in an efro.error.RemoteError being returned to the sender, the exception will be logged on the receiver. This is because the goal is usually to avoid returning opaque RemoteErrors and to instead return something meaningful as part of the expected response type (even if that value itself represents a logical error state). If 'log_errors_on_receiver' is False, however, such exceptions will *not* be logged on the receiver. This can be useful in combination with 'remote_errors_include_stack_traces' and 'forward_clean_errors' in situations where all error logging/management will be happening on the sender end. Be aware, however, that in that case it may be possible for communication errors to prevent such error messages from ever being seen.
message_types_by_id: dict[int, type[Message]]
message_ids_by_type: dict[type[Message], int]
response_types_by_id: dict[int, type[Response] | type[SysResponse]]
response_ids_by_type: dict[type[Response] | type[SysResponse], int]
@staticmethod
def
encode_dict(obj: dict) -> str:
174 @staticmethod 175 def encode_dict(obj: dict) -> str: 176 """Json-encode a provided dict.""" 177 return json.dumps(obj, separators=(',', ':'))
Json-encode a provided dict.
179 def message_to_dict(self, message: Message) -> dict: 180 """Encode a message to a json ready dict.""" 181 return self._to_dict(message, self.message_ids_by_type, 'message')
Encode a message to a json ready dict.
183 def response_to_dict(self, response: Response | SysResponse) -> dict: 184 """Encode a response to a json ready dict.""" 185 return self._to_dict(response, self.response_ids_by_type, 'response')
Encode a response to a json ready dict.
187 def error_to_response(self, exc: Exception) -> tuple[SysResponse, bool]: 188 """Translate an Exception to a SysResponse. 189 190 Also returns whether the error should be logged if this happened 191 within handle_raw_message(). 192 """ 193 194 # If anything goes wrong, return a ErrorSysResponse instead. 195 # (either CLEAN or generic REMOTE) 196 if self.forward_clean_errors and isinstance(exc, CleanError): 197 return ( 198 ErrorSysResponse( 199 error_message=str(exc), 200 error_type=ErrorSysResponse.ErrorType.REMOTE_CLEAN, 201 ), 202 False, 203 ) 204 if self.forward_communication_errors and isinstance( 205 exc, CommunicationError 206 ): 207 return ( 208 ErrorSysResponse( 209 error_message=str(exc), 210 error_type=ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION, 211 ), 212 False, 213 ) 214 return ( 215 ErrorSysResponse( 216 error_message=( 217 # Note: need to format exception ourself here; it 218 # might not be current so we can't use 219 # traceback.format_exc(). 220 ''.join( 221 traceback.format_exception( 222 type(exc), exc, exc.__traceback__ 223 ) 224 ) 225 if self.remote_errors_include_stack_traces 226 else 'An internal error has occurred.' 227 ), 228 error_type=ErrorSysResponse.ErrorType.REMOTE, 229 ), 230 self.log_errors_on_receiver, 231 )
Translate an Exception to a SysResponse.
Also returns whether the error should be logged if this happened within handle_raw_message().
@staticmethod
def
decode_dict(data: str) -> dict:
247 @staticmethod 248 def decode_dict(data: str) -> dict: 249 """Decode data to a dict.""" 250 out = json.loads(data) 251 assert isinstance(out, dict) 252 return out
Decode data to a dict.
254 def message_from_dict(self, data: dict) -> Message: 255 """Decode a message from a json string.""" 256 out = self._from_dict(data, self.message_types_by_id, 'message') 257 assert isinstance(out, Message) 258 return out
Decode a message from a json string.
260 def response_from_dict(self, data: dict) -> Response | SysResponse: 261 """Decode a response from a json string.""" 262 out = self._from_dict(data, self.response_types_by_id, 'response') 263 assert isinstance(out, Response | SysResponse) 264 return out
Decode a response from a json string.
def
do_create_sender_module( self, basename: str, protocol_create_code: str, enable_sync_sends: bool, enable_async_sends: bool, private: bool = False, protocol_module_level_import_code: str | None = None) -> str:
408 def do_create_sender_module( 409 self, 410 basename: str, 411 protocol_create_code: str, 412 enable_sync_sends: bool, 413 enable_async_sends: bool, 414 private: bool = False, 415 protocol_module_level_import_code: str | None = None, 416 ) -> str: 417 """Used by create_sender_module(); do not call directly.""" 418 # pylint: disable=too-many-locals 419 # pylint: disable=too-many-branches 420 import textwrap 421 422 msgtypes = list(self.message_ids_by_type.keys()) 423 424 ppre = '_' if private else '' 425 out = self._get_module_header( 426 'sender', 427 extra_import_code=protocol_module_level_import_code, 428 enable_async_sends=enable_async_sends, 429 ) 430 ccind = textwrap.indent(protocol_create_code, ' ') 431 out += ( 432 f'class {ppre}{basename}(MessageSender):\n' 433 f' """Protocol-specific sender."""\n' 434 f'\n' 435 f' def __init__(self) -> None:\n' 436 f'{ccind}\n' 437 f' super().__init__(protocol)\n' 438 f'\n' 439 f' def __get__(\n' 440 f' self, obj: Any, type_in: Any = None\n' 441 f' ) -> {ppre}Bound{basename}:\n' 442 f' return {ppre}Bound{basename}(obj, self)\n' 443 f'\n' 444 f'\n' 445 f'class {ppre}Bound{basename}(BoundMessageSender):\n' 446 f' """Protocol-specific bound sender."""\n' 447 ) 448 449 def _filt_tp_name(rtype: type[Response] | None) -> str: 450 return 'None' if rtype is None else rtype.__name__ 451 452 # Define handler() overloads for all registered message types. 453 if msgtypes: 454 for async_pass in False, True: 455 if async_pass and not enable_async_sends: 456 continue 457 if not async_pass and not enable_sync_sends: 458 continue 459 pfx = 'async ' if async_pass else '' 460 sfx = '_async' if async_pass else '' 461 # awt = 'await ' if async_pass else '' 462 awt = '' 463 how = 'asynchronously' if async_pass else 'synchronously' 464 465 if len(msgtypes) == 1: 466 # Special case: with a single message types we don't 467 # use overloads. 468 msgtype = msgtypes[0] 469 msgtypevar = msgtype.__name__ 470 rtypes = msgtype.get_response_types() 471 if len(rtypes) > 1: 472 rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes) 473 else: 474 rtypevar = _filt_tp_name(rtypes[0]) 475 if async_pass: 476 rtypevar = f'Awaitable[{rtypevar}]' 477 out += ( 478 f'\n' 479 f' def send{sfx}(self,' 480 f' message: {msgtypevar})' 481 f' -> {rtypevar}:\n' 482 f' """Send a message {how}."""\n' 483 f' out = {awt}self._sender.' 484 f'send{sfx}(self._obj, message)\n' 485 ) 486 if not async_pass: 487 out += ( 488 f' assert isinstance(out, {rtypevar})\n' 489 ' return out\n' 490 ) 491 else: 492 out += f' return cast({rtypevar}, out)\n' 493 494 else: 495 for msgtype in msgtypes: 496 msgtypevar = msgtype.__name__ 497 rtypes = msgtype.get_response_types() 498 if len(rtypes) > 1: 499 rtypevar = ' | '.join( 500 _filt_tp_name(t) for t in rtypes 501 ) 502 else: 503 rtypevar = _filt_tp_name(rtypes[0]) 504 out += ( 505 f'\n' 506 f' @overload\n' 507 f' {pfx}def send{sfx}(self,' 508 f' message: {msgtypevar})' 509 f' -> {rtypevar}: ...\n' 510 ) 511 rtypevar = 'Response | None' 512 if async_pass: 513 rtypevar = f'Awaitable[{rtypevar}]' 514 out += ( 515 f'\n' 516 f' def send{sfx}(self, message: Message)' 517 f' -> {rtypevar}:\n' 518 f' """Send a message {how}."""\n' 519 f' return {awt}self._sender.' 520 f'send{sfx}(self._obj, message)\n' 521 ) 522 523 return out
Used by create_sender_module(); do not call directly.
def
do_create_receiver_module( self, basename: str, protocol_create_code: str, is_async: bool, private: bool = False, protocol_module_level_import_code: str | None = None) -> str:
525 def do_create_receiver_module( 526 self, 527 basename: str, 528 protocol_create_code: str, 529 is_async: bool, 530 private: bool = False, 531 protocol_module_level_import_code: str | None = None, 532 ) -> str: 533 """Used by create_receiver_module(); do not call directly.""" 534 # pylint: disable=too-many-locals 535 import textwrap 536 537 desc = 'asynchronous' if is_async else 'synchronous' 538 ppre = '_' if private else '' 539 msgtypes = list(self.message_ids_by_type.keys()) 540 out = self._get_module_header( 541 'receiver', 542 extra_import_code=protocol_module_level_import_code, 543 enable_async_sends=False, 544 ) 545 ccind = textwrap.indent(protocol_create_code, ' ') 546 out += ( 547 f'class {ppre}{basename}(MessageReceiver):\n' 548 f' """Protocol-specific {desc} receiver."""\n' 549 f'\n' 550 f' is_async = {is_async}\n' 551 f'\n' 552 f' def __init__(self) -> None:\n' 553 f'{ccind}\n' 554 f' super().__init__(protocol)\n' 555 f'\n' 556 f' def __get__(\n' 557 f' self,\n' 558 f' obj: Any,\n' 559 f' type_in: Any = None,\n' 560 f' ) -> {ppre}Bound{basename}:\n' 561 f' return {ppre}Bound{basename}(' 562 f'obj, self)\n' 563 ) 564 565 # Define handler() overloads for all registered message types. 566 567 def _filt_tp_name(rtype: type[Response] | None) -> str: 568 return 'None' if rtype is None else rtype.__name__ 569 570 if msgtypes: 571 cbgn = 'Awaitable[' if is_async else '' 572 cend = ']' if is_async else '' 573 if len(msgtypes) == 1: 574 # Special case: when we have a single message type we don't 575 # use overloads. 576 msgtype = msgtypes[0] 577 msgtypevar = msgtype.__name__ 578 rtypes = msgtype.get_response_types() 579 if len(rtypes) > 1: 580 rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes) 581 else: 582 rtypevar = _filt_tp_name(rtypes[0]) 583 rtypevar = f'{cbgn}{rtypevar}{cend}' 584 out += ( 585 f'\n' 586 f' def handler(\n' 587 f' self,\n' 588 f' call: Callable[[Any, {msgtypevar}], ' 589 f'{rtypevar}],\n' 590 f' )' 591 f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n' 592 f' """Decorator to register message handlers."""\n' 593 f' from typing import cast, Callable, Any\n' 594 f'\n' 595 f' self.register_handler(cast(Callable' 596 f'[[Any, Message], Response], call))\n' 597 f' return call\n' 598 ) 599 else: 600 for msgtype in msgtypes: 601 msgtypevar = msgtype.__name__ 602 rtypes = msgtype.get_response_types() 603 if len(rtypes) > 1: 604 rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes) 605 else: 606 rtypevar = _filt_tp_name(rtypes[0]) 607 rtypevar = f'{cbgn}{rtypevar}{cend}' 608 out += ( 609 f'\n' 610 f' @overload\n' 611 f' def handler(\n' 612 f' self,\n' 613 f' call: Callable[[Any, {msgtypevar}], ' 614 f'{rtypevar}],\n' 615 f' )' 616 f' -> Callable[[Any, {msgtypevar}], {rtypevar}]: ...\n' 617 ) 618 out += ( 619 '\n' 620 ' def handler(self, call: Callable) -> Callable:\n' 621 ' """Decorator to register message handlers."""\n' 622 ' self.register_handler(call)\n' 623 ' return call\n' 624 ) 625 626 out += ( 627 f'\n' 628 f'\n' 629 f'class {ppre}Bound{basename}(BoundMessageReceiver):\n' 630 f' """Protocol-specific bound receiver."""\n' 631 ) 632 if is_async: 633 out += ( 634 '\n' 635 ' def handle_raw_message(\n' 636 ' self, message: str, raise_unregistered: bool = False\n' 637 ' ) -> Awaitable[str]:\n' 638 ' """Asynchronously handle a raw incoming message."""\n' 639 ' return self._receiver.' 640 'handle_raw_message_async(\n' 641 ' self._obj, message, raise_unregistered\n' 642 ' )\n' 643 ) 644 645 else: 646 out += ( 647 '\n' 648 ' def handle_raw_message(\n' 649 ' self, message: str, raise_unregistered: bool = False\n' 650 ' ) -> str:\n' 651 ' """Synchronously handle a raw incoming message."""\n' 652 ' return self._receiver.handle_raw_message(\n' 653 ' self._obj, message, raise_unregistered\n' 654 ' )\n' 655 ) 656 657 return out
Used by create_receiver_module(); do not call directly.
class
MessageSender:
22class MessageSender: 23 """Facilitates sending messages to a target and receiving responses. 24 25 These are instantiated at the class level and used to register unbound 26 class methods to handle raw message sending. Generally this class is not 27 used directly, but instead autogenerated subclasses which provide type 28 safe overloads are used instead. 29 30 Example: 31 (In this example, MyMessageSender is an autogenerated class that 32 inherits from MessageSender). 33 34 class MyClass: 35 msg = MyMessageSender() 36 37 @msg.send_method 38 def send_raw_message(self, message: str) -> str: 39 # Actually send the message here. 40 41 obj = MyClass() 42 43 # The MyMessageSender generated class would provides overloads for 44 # send(), send_async(), etc. to provide type-safety for message types 45 # and their associated response types. 46 # Thus, given the statement below, a type-checker would know that 47 # 'response' is a SomeResponseType or whatever is associated with 48 # SomeMessageType. 49 response = obj.msg.send(SomeMessageType()) 50 51 """ 52 53 def __init__(self, protocol: MessageProtocol) -> None: 54 self.protocol = protocol 55 self._send_raw_message_call: Callable[[Any, str], str] | None = None 56 self._send_raw_message_ex_call: ( 57 Callable[[Any, str, Message], str] | None 58 ) = None 59 self._send_async_raw_message_call: ( 60 Callable[[Any, str], Awaitable[str]] | None 61 ) = None 62 self._send_async_raw_message_ex_call: ( 63 Callable[[Any, str, Message], Awaitable[str]] | None 64 ) = None 65 self._encode_filter_call: ( 66 Callable[[Any, Message, dict], None] | None 67 ) = None 68 self._decode_filter_call: ( 69 Callable[[Any, Message, dict, Response | SysResponse], None] | None 70 ) = None 71 self._peer_desc_call: Callable[[Any], str] | None = None 72 73 def send_method( 74 self, call: Callable[[Any, str], str] 75 ) -> Callable[[Any, str], str]: 76 """Function decorator for setting raw send method. 77 78 Send methods take strings and should return strings. 79 CommunicationErrors raised here will be returned to the sender 80 as such; all other exceptions will result in a RuntimeError for 81 the sender. 82 """ 83 assert self._send_raw_message_call is None 84 self._send_raw_message_call = call 85 return call 86 87 def send_ex_method( 88 self, call: Callable[[Any, str, Message], str] 89 ) -> Callable[[Any, str, Message], str]: 90 """Function decorator for extended send method. 91 92 Version of send_method which is also is passed the original 93 unencoded message; can be useful for cases where metadata is sent 94 along with messages referring to their payloads/etc. 95 """ 96 assert self._send_raw_message_ex_call is None 97 self._send_raw_message_ex_call = call 98 return call 99 100 def send_async_method( 101 self, call: Callable[[Any, str], Awaitable[str]] 102 ) -> Callable[[Any, str], Awaitable[str]]: 103 """Function decorator for setting raw send-async method. 104 105 Send methods take strings and should return strings. 106 CommunicationErrors raised here will be returned to the sender 107 as such; all other exceptions will result in a RuntimeError for 108 the sender. 109 110 IMPORTANT: Generally async send methods should not be implemented 111 as 'async' methods, but instead should be regular methods that 112 return awaitable objects. This way it can be guaranteed that 113 outgoing messages are synchronously enqueued in the correct 114 order, and then async calls can be returned which finish each 115 send. If the entire call is async, they may be enqueued out of 116 order in rare cases. 117 """ 118 assert self._send_async_raw_message_call is None 119 self._send_async_raw_message_call = call 120 return call 121 122 def send_async_ex_method( 123 self, call: Callable[[Any, str, Message], Awaitable[str]] 124 ) -> Callable[[Any, str, Message], Awaitable[str]]: 125 """Function decorator for extended send-async method. 126 127 Version of send_async_method which is also is passed the original 128 unencoded message; can be useful for cases where metadata is sent 129 along with messages referring to their payloads/etc. 130 """ 131 assert self._send_async_raw_message_ex_call is None 132 self._send_async_raw_message_ex_call = call 133 return call 134 135 def encode_filter_method( 136 self, call: Callable[[Any, Message, dict], None] 137 ) -> Callable[[Any, Message, dict], None]: 138 """Function decorator for defining an encode filter. 139 140 Encode filters can be used to add extra data to the message 141 dict before is is encoded to a string and sent out. 142 """ 143 assert self._encode_filter_call is None 144 self._encode_filter_call = call 145 return call 146 147 def decode_filter_method( 148 self, call: Callable[[Any, Message, dict, Response | SysResponse], None] 149 ) -> Callable[[Any, Message, dict, Response], None]: 150 """Function decorator for defining a decode filter. 151 152 Decode filters can be used to extract extra data from incoming 153 message dicts. 154 """ 155 assert self._decode_filter_call is None 156 self._decode_filter_call = call 157 return call 158 159 def peer_desc_method( 160 self, call: Callable[[Any], str] 161 ) -> Callable[[Any], str]: 162 """Function decorator for defining peer descriptions. 163 164 These are included in error messages or other diagnostics. 165 """ 166 assert self._peer_desc_call is None 167 self._peer_desc_call = call 168 return call 169 170 def send(self, bound_obj: Any, message: Message) -> Response | None: 171 """Send a message synchronously.""" 172 return self.unpack_raw_response( 173 bound_obj=bound_obj, 174 message=message, 175 raw_response=self.fetch_raw_response( 176 bound_obj=bound_obj, 177 message=message, 178 ), 179 ) 180 181 def send_async( 182 self, bound_obj: Any, message: Message 183 ) -> Awaitable[Response | None]: 184 """Send a message asynchronously.""" 185 186 # Note: This call is synchronous so that the first part of it can 187 # happen synchronously. If the whole call were async we wouldn't be 188 # able to guarantee that messages sent in order would actually go 189 # out in order. 190 raw_response_awaitable = self.fetch_raw_response_async( 191 bound_obj=bound_obj, 192 message=message, 193 ) 194 # Now return an awaitable that will finish the send. 195 return self._send_async_awaitable( 196 bound_obj, message, raw_response_awaitable 197 ) 198 199 async def _send_async_awaitable( 200 self, 201 bound_obj: Any, 202 message: Message, 203 raw_response_awaitable: Awaitable[Response | SysResponse], 204 ) -> Response | None: 205 return self.unpack_raw_response( 206 bound_obj=bound_obj, 207 message=message, 208 raw_response=await raw_response_awaitable, 209 ) 210 211 def fetch_raw_response( 212 self, bound_obj: Any, message: Message 213 ) -> Response | SysResponse: 214 """Send a message synchronously. 215 216 Generally you can just call send(); these split versions are 217 for when message sending and response handling need to happen 218 in different contexts/threads. 219 """ 220 if ( 221 self._send_raw_message_call is None 222 and self._send_raw_message_ex_call is None 223 ): 224 raise RuntimeError('send() is unimplemented for this type.') 225 226 msg_encoded = self._encode_message(bound_obj, message) 227 try: 228 if self._send_raw_message_ex_call is not None: 229 response_encoded = self._send_raw_message_ex_call( 230 bound_obj, msg_encoded, message 231 ) 232 else: 233 assert self._send_raw_message_call is not None 234 response_encoded = self._send_raw_message_call( 235 bound_obj, msg_encoded 236 ) 237 except Exception as exc: 238 response = ErrorSysResponse( 239 error_message='Error in MessageSender @send_method.', 240 error_type=( 241 ErrorSysResponse.ErrorType.COMMUNICATION 242 if isinstance(exc, CommunicationError) 243 else ErrorSysResponse.ErrorType.LOCAL 244 ), 245 ) 246 # Can include the actual exception since we'll be looking at 247 # this locally; might be helpful. 248 response.set_local_exception(exc) 249 return response 250 return self._decode_raw_response(bound_obj, message, response_encoded) 251 252 def fetch_raw_response_async( 253 self, bound_obj: Any, message: Message 254 ) -> Awaitable[Response | SysResponse]: 255 """Fetch a raw message response awaitable. 256 257 The result of this should be awaited and then passed to 258 unpack_raw_response() to produce the final message result. 259 260 Generally you can just call send(); calling fetch and unpack 261 manually is for when message sending and response handling need 262 to happen in different contexts/threads. 263 """ 264 265 # Note: This call is synchronous so that the first part of it can 266 # happen synchronously. If the whole call were async we wouldn't be 267 # able to guarantee that messages sent in order would actually go 268 # out in order. 269 if ( 270 self._send_async_raw_message_call is None 271 and self._send_async_raw_message_ex_call is None 272 ): 273 raise RuntimeError('send_async() is unimplemented for this type.') 274 275 msg_encoded = self._encode_message(bound_obj, message) 276 try: 277 if self._send_async_raw_message_ex_call is not None: 278 send_awaitable = self._send_async_raw_message_ex_call( 279 bound_obj, msg_encoded, message 280 ) 281 else: 282 assert self._send_async_raw_message_call is not None 283 send_awaitable = self._send_async_raw_message_call( 284 bound_obj, msg_encoded 285 ) 286 except Exception as exc: 287 return self._error_awaitable(exc) 288 289 # Now return an awaitable to finish the job. 290 return self._fetch_raw_response_awaitable( 291 bound_obj, message, send_awaitable 292 ) 293 294 async def _error_awaitable(self, exc: Exception) -> SysResponse: 295 response = ErrorSysResponse( 296 error_message='Error in MessageSender @send_async_method.', 297 error_type=( 298 ErrorSysResponse.ErrorType.COMMUNICATION 299 if isinstance(exc, CommunicationError) 300 else ErrorSysResponse.ErrorType.LOCAL 301 ), 302 ) 303 # Can include the actual exception since we'll be looking at 304 # this locally; might be helpful. 305 response.set_local_exception(exc) 306 return response 307 308 async def _fetch_raw_response_awaitable( 309 self, bound_obj: Any, message: Message, send_awaitable: Awaitable[str] 310 ) -> Response | SysResponse: 311 try: 312 response_encoded = await send_awaitable 313 except Exception as exc: 314 response = ErrorSysResponse( 315 error_message='Error in MessageSender @send_async_method.', 316 error_type=( 317 ErrorSysResponse.ErrorType.COMMUNICATION 318 if isinstance(exc, CommunicationError) 319 else ErrorSysResponse.ErrorType.LOCAL 320 ), 321 ) 322 # Can include the actual exception since we'll be looking at 323 # this locally; might be helpful. 324 response.set_local_exception(exc) 325 return response 326 return self._decode_raw_response(bound_obj, message, response_encoded) 327 328 def unpack_raw_response( 329 self, 330 bound_obj: Any, 331 message: Message, 332 raw_response: Response | SysResponse, 333 ) -> Response | None: 334 """Convert a raw fetched response into a final response/error/etc. 335 336 Generally you can just call send(); calling fetch and unpack 337 manually is for when message sending and response handling need 338 to happen in different contexts/threads. 339 """ 340 response = self._unpack_raw_response(bound_obj, raw_response) 341 assert ( 342 response is None 343 or type(response) in type(message).get_response_types() 344 ) 345 return response 346 347 def _encode_message(self, bound_obj: Any, message: Message) -> str: 348 """Encode a message for sending.""" 349 msg_dict = self.protocol.message_to_dict(message) 350 if self._encode_filter_call is not None: 351 self._encode_filter_call(bound_obj, message, msg_dict) 352 return self.protocol.encode_dict(msg_dict) 353 354 def _decode_raw_response( 355 self, bound_obj: Any, message: Message, response_encoded: str 356 ) -> Response | SysResponse: 357 """Create a Response from returned data. 358 359 These Responses may encapsulate things like remote errors and 360 should not be handed directly to users. _unpack_raw_response() 361 should be used to translate to special values like None or raise 362 Exceptions. This function itself should never raise Exceptions. 363 """ 364 response: Response | SysResponse 365 try: 366 response_dict = self.protocol.decode_dict(response_encoded) 367 response = self.protocol.response_from_dict(response_dict) 368 if self._decode_filter_call is not None: 369 self._decode_filter_call( 370 bound_obj, message, response_dict, response 371 ) 372 except Exception as exc: 373 response = ErrorSysResponse( 374 error_message='Error decoding raw response.', 375 error_type=ErrorSysResponse.ErrorType.LOCAL, 376 ) 377 # Since we'll be looking at this locally, we can include 378 # extra info for logging/etc. 379 response.set_local_exception(exc) 380 return response 381 382 def _unpack_raw_response( 383 self, bound_obj: Any, raw_response: Response | SysResponse 384 ) -> Response | None: 385 """Given a raw Response, unpacks to special values or Exceptions. 386 387 The result of this call is what should be passed to users. 388 For complex messaging situations such as response callbacks 389 operating across different threads, this last stage should be 390 run such that any raised Exception is active when the callback 391 fires; not on the thread where the message was sent. 392 """ 393 # EmptySysResponse translates to None 394 if isinstance(raw_response, EmptySysResponse): 395 return None 396 397 # Some error occurred. Raise a local Exception for it. 398 if isinstance(raw_response, ErrorSysResponse): 399 # Errors that happened locally can attach their exceptions 400 # here for extra logging goodness. 401 local_exception = raw_response.get_local_exception() 402 403 if ( 404 raw_response.error_type 405 is ErrorSysResponse.ErrorType.COMMUNICATION 406 ): 407 raise CommunicationError( 408 raw_response.error_message 409 ) from local_exception 410 411 # If something went wrong on *our* end of the connection, 412 # don't say it was a remote error. 413 if raw_response.error_type is ErrorSysResponse.ErrorType.LOCAL: 414 raise RuntimeError( 415 raw_response.error_message 416 ) from local_exception 417 418 # If they want to support clean errors, do those. 419 if ( 420 self.protocol.forward_clean_errors 421 and raw_response.error_type 422 is ErrorSysResponse.ErrorType.REMOTE_CLEAN 423 ): 424 raise CleanError( 425 raw_response.error_message 426 ) from local_exception 427 428 if ( 429 self.protocol.forward_communication_errors 430 and raw_response.error_type 431 is ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION 432 ): 433 raise CommunicationError( 434 raw_response.error_message 435 ) from local_exception 436 437 # Everything else gets lumped in as a remote error. 438 raise RemoteError( 439 raw_response.error_message, 440 peer_desc=( 441 'peer' 442 if self._peer_desc_call is None 443 else self._peer_desc_call(bound_obj) 444 ), 445 ) from local_exception 446 447 assert isinstance(raw_response, Response) 448 return raw_response
Facilitates sending messages to a target and receiving responses.
These are instantiated at the class level and used to register unbound class methods to handle raw message sending. Generally this class is not used directly, but instead autogenerated subclasses which provide type safe overloads are used instead.
Example: (In this example, MyMessageSender is an autogenerated class that inherits from MessageSender).
class MyClass: msg = MyMessageSender()
@msg.send_method
def send_raw_message(self, message: str) -> str:
# Actually send the message here.
obj = MyClass()
The MyMessageSender generated class would provides overloads for
send(), send_async(), etc. to provide type-safety for message types
and their associated response types.
Thus, given the statement below, a type-checker would know that
'response' is a SomeResponseType or whatever is associated with
SomeMessageType.
response = obj.msg.send(SomeMessageType())
MessageSender(protocol: MessageProtocol)
53 def __init__(self, protocol: MessageProtocol) -> None: 54 self.protocol = protocol 55 self._send_raw_message_call: Callable[[Any, str], str] | None = None 56 self._send_raw_message_ex_call: ( 57 Callable[[Any, str, Message], str] | None 58 ) = None 59 self._send_async_raw_message_call: ( 60 Callable[[Any, str], Awaitable[str]] | None 61 ) = None 62 self._send_async_raw_message_ex_call: ( 63 Callable[[Any, str, Message], Awaitable[str]] | None 64 ) = None 65 self._encode_filter_call: ( 66 Callable[[Any, Message, dict], None] | None 67 ) = None 68 self._decode_filter_call: ( 69 Callable[[Any, Message, dict, Response | SysResponse], None] | None 70 ) = None 71 self._peer_desc_call: Callable[[Any], str] | None = None
def
send_method(self, call: Callable[[Any, str], str]) -> Callable[[Any, str], str]:
73 def send_method( 74 self, call: Callable[[Any, str], str] 75 ) -> Callable[[Any, str], str]: 76 """Function decorator for setting raw send method. 77 78 Send methods take strings and should return strings. 79 CommunicationErrors raised here will be returned to the sender 80 as such; all other exceptions will result in a RuntimeError for 81 the sender. 82 """ 83 assert self._send_raw_message_call is None 84 self._send_raw_message_call = call 85 return call
Function decorator for setting raw send method.
Send methods take strings and should return strings. CommunicationErrors raised here will be returned to the sender as such; all other exceptions will result in a RuntimeError for the sender.
def
send_ex_method( self, call: Callable[[Any, str, Message], str]) -> Callable[[Any, str, Message], str]:
87 def send_ex_method( 88 self, call: Callable[[Any, str, Message], str] 89 ) -> Callable[[Any, str, Message], str]: 90 """Function decorator for extended send method. 91 92 Version of send_method which is also is passed the original 93 unencoded message; can be useful for cases where metadata is sent 94 along with messages referring to their payloads/etc. 95 """ 96 assert self._send_raw_message_ex_call is None 97 self._send_raw_message_ex_call = call 98 return call
Function decorator for extended send method.
Version of send_method which is also is passed the original unencoded message; can be useful for cases where metadata is sent along with messages referring to their payloads/etc.
def
send_async_method( self, call: Callable[[Any, str], Awaitable[str]]) -> Callable[[Any, str], Awaitable[str]]:
100 def send_async_method( 101 self, call: Callable[[Any, str], Awaitable[str]] 102 ) -> Callable[[Any, str], Awaitable[str]]: 103 """Function decorator for setting raw send-async method. 104 105 Send methods take strings and should return strings. 106 CommunicationErrors raised here will be returned to the sender 107 as such; all other exceptions will result in a RuntimeError for 108 the sender. 109 110 IMPORTANT: Generally async send methods should not be implemented 111 as 'async' methods, but instead should be regular methods that 112 return awaitable objects. This way it can be guaranteed that 113 outgoing messages are synchronously enqueued in the correct 114 order, and then async calls can be returned which finish each 115 send. If the entire call is async, they may be enqueued out of 116 order in rare cases. 117 """ 118 assert self._send_async_raw_message_call is None 119 self._send_async_raw_message_call = call 120 return call
Function decorator for setting raw send-async method.
Send methods take strings and should return strings. CommunicationErrors raised here will be returned to the sender as such; all other exceptions will result in a RuntimeError for the sender.
IMPORTANT: Generally async send methods should not be implemented as 'async' methods, but instead should be regular methods that return awaitable objects. This way it can be guaranteed that outgoing messages are synchronously enqueued in the correct order, and then async calls can be returned which finish each send. If the entire call is async, they may be enqueued out of order in rare cases.
def
send_async_ex_method( self, call: Callable[[Any, str, Message], Awaitable[str]]) -> Callable[[Any, str, Message], Awaitable[str]]:
122 def send_async_ex_method( 123 self, call: Callable[[Any, str, Message], Awaitable[str]] 124 ) -> Callable[[Any, str, Message], Awaitable[str]]: 125 """Function decorator for extended send-async method. 126 127 Version of send_async_method which is also is passed the original 128 unencoded message; can be useful for cases where metadata is sent 129 along with messages referring to their payloads/etc. 130 """ 131 assert self._send_async_raw_message_ex_call is None 132 self._send_async_raw_message_ex_call = call 133 return call
Function decorator for extended send-async method.
Version of send_async_method which is also is passed the original unencoded message; can be useful for cases where metadata is sent along with messages referring to their payloads/etc.
def
encode_filter_method( self, call: Callable[[Any, Message, dict], NoneType]) -> Callable[[Any, Message, dict], NoneType]:
135 def encode_filter_method( 136 self, call: Callable[[Any, Message, dict], None] 137 ) -> Callable[[Any, Message, dict], None]: 138 """Function decorator for defining an encode filter. 139 140 Encode filters can be used to add extra data to the message 141 dict before is is encoded to a string and sent out. 142 """ 143 assert self._encode_filter_call is None 144 self._encode_filter_call = call 145 return call
Function decorator for defining an encode filter.
Encode filters can be used to add extra data to the message dict before is is encoded to a string and sent out.
def
decode_filter_method( self, call: Callable[[Any, Message, dict, Response | SysResponse], NoneType]) -> Callable[[Any, Message, dict, Response], NoneType]:
147 def decode_filter_method( 148 self, call: Callable[[Any, Message, dict, Response | SysResponse], None] 149 ) -> Callable[[Any, Message, dict, Response], None]: 150 """Function decorator for defining a decode filter. 151 152 Decode filters can be used to extract extra data from incoming 153 message dicts. 154 """ 155 assert self._decode_filter_call is None 156 self._decode_filter_call = call 157 return call
Function decorator for defining a decode filter.
Decode filters can be used to extract extra data from incoming message dicts.
def
peer_desc_method(self, call: Callable[[Any], str]) -> Callable[[Any], str]:
159 def peer_desc_method( 160 self, call: Callable[[Any], str] 161 ) -> Callable[[Any], str]: 162 """Function decorator for defining peer descriptions. 163 164 These are included in error messages or other diagnostics. 165 """ 166 assert self._peer_desc_call is None 167 self._peer_desc_call = call 168 return call
Function decorator for defining peer descriptions.
These are included in error messages or other diagnostics.
170 def send(self, bound_obj: Any, message: Message) -> Response | None: 171 """Send a message synchronously.""" 172 return self.unpack_raw_response( 173 bound_obj=bound_obj, 174 message=message, 175 raw_response=self.fetch_raw_response( 176 bound_obj=bound_obj, 177 message=message, 178 ), 179 )
Send a message synchronously.
181 def send_async( 182 self, bound_obj: Any, message: Message 183 ) -> Awaitable[Response | None]: 184 """Send a message asynchronously.""" 185 186 # Note: This call is synchronous so that the first part of it can 187 # happen synchronously. If the whole call were async we wouldn't be 188 # able to guarantee that messages sent in order would actually go 189 # out in order. 190 raw_response_awaitable = self.fetch_raw_response_async( 191 bound_obj=bound_obj, 192 message=message, 193 ) 194 # Now return an awaitable that will finish the send. 195 return self._send_async_awaitable( 196 bound_obj, message, raw_response_awaitable 197 )
Send a message asynchronously.
211 def fetch_raw_response( 212 self, bound_obj: Any, message: Message 213 ) -> Response | SysResponse: 214 """Send a message synchronously. 215 216 Generally you can just call send(); these split versions are 217 for when message sending and response handling need to happen 218 in different contexts/threads. 219 """ 220 if ( 221 self._send_raw_message_call is None 222 and self._send_raw_message_ex_call is None 223 ): 224 raise RuntimeError('send() is unimplemented for this type.') 225 226 msg_encoded = self._encode_message(bound_obj, message) 227 try: 228 if self._send_raw_message_ex_call is not None: 229 response_encoded = self._send_raw_message_ex_call( 230 bound_obj, msg_encoded, message 231 ) 232 else: 233 assert self._send_raw_message_call is not None 234 response_encoded = self._send_raw_message_call( 235 bound_obj, msg_encoded 236 ) 237 except Exception as exc: 238 response = ErrorSysResponse( 239 error_message='Error in MessageSender @send_method.', 240 error_type=( 241 ErrorSysResponse.ErrorType.COMMUNICATION 242 if isinstance(exc, CommunicationError) 243 else ErrorSysResponse.ErrorType.LOCAL 244 ), 245 ) 246 # Can include the actual exception since we'll be looking at 247 # this locally; might be helpful. 248 response.set_local_exception(exc) 249 return response 250 return self._decode_raw_response(bound_obj, message, response_encoded)
Send a message synchronously.
Generally you can just call send(); these split versions are for when message sending and response handling need to happen in different contexts/threads.
def
fetch_raw_response_async( self, bound_obj: Any, message: Message) -> Awaitable[Response | SysResponse]:
252 def fetch_raw_response_async( 253 self, bound_obj: Any, message: Message 254 ) -> Awaitable[Response | SysResponse]: 255 """Fetch a raw message response awaitable. 256 257 The result of this should be awaited and then passed to 258 unpack_raw_response() to produce the final message result. 259 260 Generally you can just call send(); calling fetch and unpack 261 manually is for when message sending and response handling need 262 to happen in different contexts/threads. 263 """ 264 265 # Note: This call is synchronous so that the first part of it can 266 # happen synchronously. If the whole call were async we wouldn't be 267 # able to guarantee that messages sent in order would actually go 268 # out in order. 269 if ( 270 self._send_async_raw_message_call is None 271 and self._send_async_raw_message_ex_call is None 272 ): 273 raise RuntimeError('send_async() is unimplemented for this type.') 274 275 msg_encoded = self._encode_message(bound_obj, message) 276 try: 277 if self._send_async_raw_message_ex_call is not None: 278 send_awaitable = self._send_async_raw_message_ex_call( 279 bound_obj, msg_encoded, message 280 ) 281 else: 282 assert self._send_async_raw_message_call is not None 283 send_awaitable = self._send_async_raw_message_call( 284 bound_obj, msg_encoded 285 ) 286 except Exception as exc: 287 return self._error_awaitable(exc) 288 289 # Now return an awaitable to finish the job. 290 return self._fetch_raw_response_awaitable( 291 bound_obj, message, send_awaitable 292 )
Fetch a raw message response awaitable.
The result of this should be awaited and then passed to unpack_raw_response() to produce the final message result.
Generally you can just call send(); calling fetch and unpack manually is for when message sending and response handling need to happen in different contexts/threads.
def
unpack_raw_response( self, bound_obj: Any, message: Message, raw_response: Response | SysResponse) -> Response | None:
328 def unpack_raw_response( 329 self, 330 bound_obj: Any, 331 message: Message, 332 raw_response: Response | SysResponse, 333 ) -> Response | None: 334 """Convert a raw fetched response into a final response/error/etc. 335 336 Generally you can just call send(); calling fetch and unpack 337 manually is for when message sending and response handling need 338 to happen in different contexts/threads. 339 """ 340 response = self._unpack_raw_response(bound_obj, raw_response) 341 assert ( 342 response is None 343 or type(response) in type(message).get_response_types() 344 ) 345 return response
Convert a raw fetched response into a final response/error/etc.
Generally you can just call send(); calling fetch and unpack manually is for when message sending and response handling need to happen in different contexts/threads.
class
BoundMessageSender:
451class BoundMessageSender: 452 """Base class for bound senders.""" 453 454 def __init__(self, obj: Any, sender: MessageSender) -> None: 455 # Note: not checking obj here since we want to support 456 # at least our protocol property when accessed via type. 457 self._obj = obj 458 self._sender = sender 459 460 @property 461 def protocol(self) -> MessageProtocol: 462 """Protocol associated with this sender.""" 463 return self._sender.protocol 464 465 def send_untyped(self, message: Message) -> Response | None: 466 """Send a message synchronously. 467 468 Whenever possible, use the send() call provided by generated 469 subclasses instead of this; it will provide better type safety. 470 """ 471 assert self._obj is not None 472 return self._sender.send(bound_obj=self._obj, message=message) 473 474 def send_async_untyped( 475 self, message: Message 476 ) -> Awaitable[Response | None]: 477 """Send a message asynchronously. 478 479 Whenever possible, use the send_async() call provided by generated 480 subclasses instead of this; it will provide better type safety. 481 """ 482 assert self._obj is not None 483 return self._sender.send_async(bound_obj=self._obj, message=message) 484 485 def fetch_raw_response_async_untyped( 486 self, message: Message 487 ) -> Awaitable[Response | SysResponse]: 488 """Split send (part 1 of 2).""" 489 assert self._obj is not None 490 return self._sender.fetch_raw_response_async( 491 bound_obj=self._obj, message=message 492 ) 493 494 def unpack_raw_response_untyped( 495 self, message: Message, raw_response: Response | SysResponse 496 ) -> Response | None: 497 """Split send (part 2 of 2).""" 498 return self._sender.unpack_raw_response( 499 bound_obj=self._obj, message=message, raw_response=raw_response 500 )
Base class for bound senders.
BoundMessageSender(obj: Any, sender: MessageSender)
protocol: MessageProtocol
460 @property 461 def protocol(self) -> MessageProtocol: 462 """Protocol associated with this sender.""" 463 return self._sender.protocol
Protocol associated with this sender.
465 def send_untyped(self, message: Message) -> Response | None: 466 """Send a message synchronously. 467 468 Whenever possible, use the send() call provided by generated 469 subclasses instead of this; it will provide better type safety. 470 """ 471 assert self._obj is not None 472 return self._sender.send(bound_obj=self._obj, message=message)
Send a message synchronously.
Whenever possible, use the send() call provided by generated subclasses instead of this; it will provide better type safety.
474 def send_async_untyped( 475 self, message: Message 476 ) -> Awaitable[Response | None]: 477 """Send a message asynchronously. 478 479 Whenever possible, use the send_async() call provided by generated 480 subclasses instead of this; it will provide better type safety. 481 """ 482 assert self._obj is not None 483 return self._sender.send_async(bound_obj=self._obj, message=message)
Send a message asynchronously.
Whenever possible, use the send_async() call provided by generated subclasses instead of this; it will provide better type safety.
485 def fetch_raw_response_async_untyped( 486 self, message: Message 487 ) -> Awaitable[Response | SysResponse]: 488 """Split send (part 1 of 2).""" 489 assert self._obj is not None 490 return self._sender.fetch_raw_response_async( 491 bound_obj=self._obj, message=message 492 )
Split send (part 1 of 2).
def
unpack_raw_response_untyped( self, message: Message, raw_response: Response | SysResponse) -> Response | None:
494 def unpack_raw_response_untyped( 495 self, message: Message, raw_response: Response | SysResponse 496 ) -> Response | None: 497 """Split send (part 2 of 2).""" 498 return self._sender.unpack_raw_response( 499 bound_obj=self._obj, message=message, raw_response=raw_response 500 )
Split send (part 2 of 2).
class
MessageReceiver:
29class MessageReceiver: 30 """Facilitates receiving & responding to messages from a remote source. 31 32 This is instantiated at the class level with unbound methods registered 33 as handlers for different message types in the protocol. 34 35 Example: 36 37 class MyClass: 38 receiver = MyMessageReceiver() 39 40 # MyMessageReceiver fills out handler() overloads to ensure all 41 # registered handlers have valid types/return-types. 42 43 @receiver.handler 44 def handle_some_message_type(self, message: SomeMsg) -> SomeResponse: 45 # Deal with this message type here. 46 47 # This will trigger the registered handler being called. 48 obj = MyClass() 49 obj.receiver.handle_raw_message(some_raw_data) 50 51 Any unhandled Exception occurring during message handling will result in 52 an efro.error.RemoteError being raised on the sending end. 53 """ 54 55 is_async = False 56 57 def __init__(self, protocol: MessageProtocol) -> None: 58 self.protocol = protocol 59 self._handlers: dict[type[Message], Callable] = {} 60 self._decode_filter_call: ( 61 Callable[[Any, dict, Message], None] | None 62 ) = None 63 self._encode_filter_call: ( 64 Callable[[Any, Message | None, Response | SysResponse, dict], None] 65 | None 66 ) = None 67 68 # noinspection PyProtectedMember 69 def register_handler( 70 self, call: Callable[[Any, Message], Response | None] 71 ) -> None: 72 """Register a handler call. 73 74 The message type handled by the call is determined by its 75 type annotation. 76 """ 77 # TODO: can use types.GenericAlias in 3.9. 78 # (hmm though now that we're there, it seems a drop-in 79 # replace gives us errors. Should re-test in 3.11 as it seems 80 # that typing_extensions handles it differently in that case) 81 from typing import _GenericAlias # type: ignore 82 from typing import get_type_hints, get_args 83 84 sig = inspect.getfullargspec(call) 85 86 # The provided callable should be a method taking one 'msg' arg. 87 expectedsig = ['self', 'msg'] 88 if sig.args != expectedsig: 89 raise ValueError( 90 f'Expected callable signature of {expectedsig};' 91 f' got {sig.args}' 92 ) 93 94 # Check annotation types to determine what message types we handle. 95 # Return-type annotation can be a Union, but we probably don't 96 # have it available at runtime. Explicitly pull it in. 97 # UPDATE: we've updated our pylint filter to where we should 98 # have all annotations available. 99 # anns = get_type_hints(call, localns={'Union': Union}) 100 anns = get_type_hints(call) 101 102 msgtype = anns.get('msg') 103 if not isinstance(msgtype, type): 104 raise TypeError( 105 f'expected a type for "msg" annotation; got {type(msgtype)}.' 106 ) 107 assert issubclass(msgtype, Message) 108 109 ret = anns.get('return') 110 responsetypes: tuple[type[Any] | None, ...] 111 112 # Return types can be a single type or a union of types. 113 if isinstance(ret, (_GenericAlias, types.UnionType)): 114 targs = get_args(ret) 115 if not all(isinstance(a, (type, type(None))) for a in targs): 116 raise TypeError( 117 f'expected only types for "return" annotation;' 118 f' got {targs}.' 119 ) 120 responsetypes = targs 121 else: 122 if not isinstance(ret, (type, type(None))): 123 raise TypeError( 124 f'expected one or more types for' 125 f' "return" annotation; got a {type(ret)}.' 126 ) 127 # This seems like maybe a mypy bug. Appeared after adding 128 # types.UnionType above. 129 responsetypes = (ret,) 130 131 # This will contain NoneType for empty return cases, but 132 # we expect it to be None. 133 # noinspection PyPep8 134 responsetypes = tuple( 135 None if r is type(None) else r for r in responsetypes 136 ) 137 138 # Make sure our protocol has this message type registered and our 139 # return types exactly match. (Technically we could return a subset 140 # of the supported types; can allow this in the future if it makes 141 # sense). 142 registered_types = self.protocol.message_ids_by_type.keys() 143 144 if msgtype not in registered_types: 145 raise TypeError( 146 f'Message type {msgtype} is not registered' 147 f' in this Protocol.' 148 ) 149 150 if msgtype in self._handlers: 151 raise TypeError( 152 f'Message type {msgtype} already has a registered handler.' 153 ) 154 155 # Make sure the responses exactly matches what the message expects. 156 if set(responsetypes) != set(msgtype.get_response_types()): 157 raise TypeError( 158 f'Provided response types {responsetypes} do not' 159 f' match the set expected by message type {msgtype}: ' 160 f'({msgtype.get_response_types()})' 161 ) 162 163 # Ok; we're good! 164 self._handlers[msgtype] = call 165 166 def decode_filter_method( 167 self, call: Callable[[Any, dict, Message], None] 168 ) -> Callable[[Any, dict, Message], None]: 169 """Function decorator for defining a decode filter. 170 171 Decode filters can be used to extract extra data from incoming 172 message dicts. This version will work for both handle_raw_message() 173 and handle_raw_message_async() 174 """ 175 assert self._decode_filter_call is None 176 self._decode_filter_call = call 177 return call 178 179 def encode_filter_method( 180 self, 181 call: Callable[ 182 [Any, Message | None, Response | SysResponse, dict], None 183 ], 184 ) -> Callable[[Any, Message | None, Response, dict], None]: 185 """Function decorator for defining an encode filter. 186 187 Encode filters can be used to add extra data to the message 188 dict before is is encoded to a string and sent out. 189 """ 190 assert self._encode_filter_call is None 191 self._encode_filter_call = call 192 return call 193 194 def validate(self, log_only: bool = False) -> None: 195 """Check for handler completeness, valid types, etc.""" 196 for msgtype in self.protocol.message_ids_by_type.keys(): 197 if issubclass(msgtype, Response): 198 continue 199 if msgtype not in self._handlers: 200 msg = ( 201 f'Protocol message type {msgtype} is not handled' 202 f' by receiver type {type(self)}.' 203 ) 204 if log_only: 205 logging.error(msg) 206 else: 207 raise TypeError(msg) 208 209 def _decode_incoming_message_base( 210 self, bound_obj: Any, msg: str 211 ) -> tuple[Any, dict, Message]: 212 # Decode the incoming message. 213 msg_dict = self.protocol.decode_dict(msg) 214 msg_decoded = self.protocol.message_from_dict(msg_dict) 215 assert isinstance(msg_decoded, Message) 216 if self._decode_filter_call is not None: 217 self._decode_filter_call(bound_obj, msg_dict, msg_decoded) 218 return bound_obj, msg_dict, msg_decoded 219 220 def _decode_incoming_message(self, bound_obj: Any, msg: str) -> Message: 221 bound_obj, _msg_dict, msg_decoded = self._decode_incoming_message_base( 222 bound_obj=bound_obj, msg=msg 223 ) 224 return msg_decoded 225 226 def encode_user_response( 227 self, bound_obj: Any, message: Message, response: Response | None 228 ) -> str: 229 """Encode a response provided by the user for sending.""" 230 231 assert isinstance(response, Response | None) 232 # (user should never explicitly return error-responses) 233 assert ( 234 response is None or type(response) in message.get_response_types() 235 ) 236 237 # A return value of None equals EmptySysResponse. 238 out_response: Response | SysResponse 239 if response is None: 240 out_response = EmptySysResponse() 241 else: 242 out_response = response 243 244 response_dict = self.protocol.response_to_dict(out_response) 245 if self._encode_filter_call is not None: 246 self._encode_filter_call( 247 bound_obj, message, out_response, response_dict 248 ) 249 return self.protocol.encode_dict(response_dict) 250 251 def encode_error_response( 252 self, bound_obj: Any, message: Message | None, exc: Exception 253 ) -> tuple[str, bool]: 254 """Given an error, return sysresponse str and whether to log.""" 255 response, dolog = self.protocol.error_to_response(exc) 256 response_dict = self.protocol.response_to_dict(response) 257 if self._encode_filter_call is not None: 258 self._encode_filter_call( 259 bound_obj, message, response, response_dict 260 ) 261 return self.protocol.encode_dict(response_dict), dolog 262 263 def handle_raw_message( 264 self, bound_obj: Any, msg: str, raise_unregistered: bool = False 265 ) -> str: 266 """Decode, handle, and return an response for a message. 267 268 if 'raise_unregistered' is True, will raise an 269 efro.message.UnregisteredMessageIDError for messages not handled by 270 the protocol. In all other cases local errors will translate to 271 error responses returned to the sender. 272 """ 273 assert not self.is_async, "can't call sync handler on async receiver" 274 msg_decoded: Message | None = None 275 try: 276 msg_decoded = self._decode_incoming_message(bound_obj, msg) 277 msgtype = type(msg_decoded) 278 handler = self._handlers.get(msgtype) 279 if handler is None: 280 raise RuntimeError(f'Got unhandled message type: {msgtype}.') 281 response = handler(bound_obj, msg_decoded) 282 assert isinstance(response, Response | None) 283 return self.encode_user_response(bound_obj, msg_decoded, response) 284 285 except Exception as exc: 286 if raise_unregistered and isinstance( 287 exc, UnregisteredMessageIDError 288 ): 289 raise 290 rstr, dolog = self.encode_error_response( 291 bound_obj, msg_decoded, exc 292 ) 293 if dolog: 294 if msg_decoded is not None: 295 msgtype = type(msg_decoded) 296 logging.exception( 297 'Error handling %s.%s message.', 298 msgtype.__module__, 299 msgtype.__qualname__, 300 ) 301 else: 302 logging.exception( 303 'Error handling raw efro.message' 304 ' (likely a message format incompatibility): %s.', 305 msg, 306 ) 307 return rstr 308 309 def handle_raw_message_async( 310 self, bound_obj: Any, msg: str, raise_unregistered: bool = False 311 ) -> Awaitable[str]: 312 """Should be called when the receiver gets a message. 313 314 The return value is the raw response to the message. 315 """ 316 317 # Note: This call is synchronous so that the first part of it can 318 # happen synchronously. If the whole call were async we wouldn't be 319 # able to guarantee that messages handlers would be called in the 320 # order the messages were received. 321 322 assert self.is_async, "Can't call async handler on sync receiver." 323 msg_decoded: Message | None = None 324 try: 325 msg_decoded = self._decode_incoming_message(bound_obj, msg) 326 msgtype = type(msg_decoded) 327 handler = self._handlers.get(msgtype) 328 if handler is None: 329 raise RuntimeError(f'Got unhandled message type: {msgtype}.') 330 handler_awaitable = handler(bound_obj, msg_decoded) 331 332 except Exception as exc: 333 if raise_unregistered and isinstance( 334 exc, UnregisteredMessageIDError 335 ): 336 raise 337 return self._handle_raw_message_async_error( 338 bound_obj, msg, msg_decoded, exc 339 ) 340 341 # Return an awaitable to handle the rest asynchronously. 342 return self._handle_raw_message_async( 343 bound_obj, msg, msg_decoded, handler_awaitable 344 ) 345 346 async def _handle_raw_message_async_error( 347 self, 348 bound_obj: Any, 349 msg_raw: str, 350 msg_decoded: Message | None, 351 exc: Exception, 352 ) -> str: 353 rstr, dolog = self.encode_error_response(bound_obj, msg_decoded, exc) 354 if dolog: 355 if msg_decoded is not None: 356 msgtype = type(msg_decoded) 357 logging.exception( 358 'Error handling %s.%s message.', 359 msgtype.__module__, 360 msgtype.__qualname__, 361 # We need to explicitly provide the exception here, 362 # otherwise it shows up at None. I assume related to 363 # the fact that we're an async function. 364 exc_info=exc, 365 ) 366 else: 367 logging.exception( 368 'Error handling raw async efro.message' 369 ' (likely a message format incompatibility): %s.', 370 msg_raw, 371 # We need to explicitly provide the exception here, 372 # otherwise it shows up at None. I assume related to 373 # the fact that we're an async function. 374 exc_info=exc, 375 ) 376 return rstr 377 378 async def _handle_raw_message_async( 379 self, 380 bound_obj: Any, 381 msg_raw: str, 382 msg_decoded: Message, 383 handler_awaitable: Awaitable[Response | None], 384 ) -> str: 385 """Should be called when the receiver gets a message. 386 387 The return value is the raw response to the message. 388 """ 389 try: 390 response = await handler_awaitable 391 assert isinstance(response, Response | None) 392 return self.encode_user_response(bound_obj, msg_decoded, response) 393 394 except Exception as exc: 395 return await self._handle_raw_message_async_error( 396 bound_obj, msg_raw, msg_decoded, exc 397 )
Facilitates receiving & responding to messages from a remote source.
This is instantiated at the class level with unbound methods registered as handlers for different message types in the protocol.
Example:
class MyClass: receiver = MyMessageReceiver()
# MyMessageReceiver fills out handler() overloads to ensure all
# registered handlers have valid types/return-types.
@receiver.handler
def handle_some_message_type(self, message: SomeMsg) -> SomeResponse:
# Deal with this message type here.
This will trigger the registered handler being called.
obj = MyClass() obj.receiver.handle_raw_message(some_raw_data)
Any unhandled Exception occurring during message handling will result in an efro.error.RemoteError being raised on the sending end.
MessageReceiver(protocol: MessageProtocol)
57 def __init__(self, protocol: MessageProtocol) -> None: 58 self.protocol = protocol 59 self._handlers: dict[type[Message], Callable] = {} 60 self._decode_filter_call: ( 61 Callable[[Any, dict, Message], None] | None 62 ) = None 63 self._encode_filter_call: ( 64 Callable[[Any, Message | None, Response | SysResponse, dict], None] 65 | None 66 ) = None
69 def register_handler( 70 self, call: Callable[[Any, Message], Response | None] 71 ) -> None: 72 """Register a handler call. 73 74 The message type handled by the call is determined by its 75 type annotation. 76 """ 77 # TODO: can use types.GenericAlias in 3.9. 78 # (hmm though now that we're there, it seems a drop-in 79 # replace gives us errors. Should re-test in 3.11 as it seems 80 # that typing_extensions handles it differently in that case) 81 from typing import _GenericAlias # type: ignore 82 from typing import get_type_hints, get_args 83 84 sig = inspect.getfullargspec(call) 85 86 # The provided callable should be a method taking one 'msg' arg. 87 expectedsig = ['self', 'msg'] 88 if sig.args != expectedsig: 89 raise ValueError( 90 f'Expected callable signature of {expectedsig};' 91 f' got {sig.args}' 92 ) 93 94 # Check annotation types to determine what message types we handle. 95 # Return-type annotation can be a Union, but we probably don't 96 # have it available at runtime. Explicitly pull it in. 97 # UPDATE: we've updated our pylint filter to where we should 98 # have all annotations available. 99 # anns = get_type_hints(call, localns={'Union': Union}) 100 anns = get_type_hints(call) 101 102 msgtype = anns.get('msg') 103 if not isinstance(msgtype, type): 104 raise TypeError( 105 f'expected a type for "msg" annotation; got {type(msgtype)}.' 106 ) 107 assert issubclass(msgtype, Message) 108 109 ret = anns.get('return') 110 responsetypes: tuple[type[Any] | None, ...] 111 112 # Return types can be a single type or a union of types. 113 if isinstance(ret, (_GenericAlias, types.UnionType)): 114 targs = get_args(ret) 115 if not all(isinstance(a, (type, type(None))) for a in targs): 116 raise TypeError( 117 f'expected only types for "return" annotation;' 118 f' got {targs}.' 119 ) 120 responsetypes = targs 121 else: 122 if not isinstance(ret, (type, type(None))): 123 raise TypeError( 124 f'expected one or more types for' 125 f' "return" annotation; got a {type(ret)}.' 126 ) 127 # This seems like maybe a mypy bug. Appeared after adding 128 # types.UnionType above. 129 responsetypes = (ret,) 130 131 # This will contain NoneType for empty return cases, but 132 # we expect it to be None. 133 # noinspection PyPep8 134 responsetypes = tuple( 135 None if r is type(None) else r for r in responsetypes 136 ) 137 138 # Make sure our protocol has this message type registered and our 139 # return types exactly match. (Technically we could return a subset 140 # of the supported types; can allow this in the future if it makes 141 # sense). 142 registered_types = self.protocol.message_ids_by_type.keys() 143 144 if msgtype not in registered_types: 145 raise TypeError( 146 f'Message type {msgtype} is not registered' 147 f' in this Protocol.' 148 ) 149 150 if msgtype in self._handlers: 151 raise TypeError( 152 f'Message type {msgtype} already has a registered handler.' 153 ) 154 155 # Make sure the responses exactly matches what the message expects. 156 if set(responsetypes) != set(msgtype.get_response_types()): 157 raise TypeError( 158 f'Provided response types {responsetypes} do not' 159 f' match the set expected by message type {msgtype}: ' 160 f'({msgtype.get_response_types()})' 161 ) 162 163 # Ok; we're good! 164 self._handlers[msgtype] = call
Register a handler call.
The message type handled by the call is determined by its type annotation.
def
decode_filter_method( self, call: Callable[[Any, dict, Message], NoneType]) -> Callable[[Any, dict, Message], NoneType]:
166 def decode_filter_method( 167 self, call: Callable[[Any, dict, Message], None] 168 ) -> Callable[[Any, dict, Message], None]: 169 """Function decorator for defining a decode filter. 170 171 Decode filters can be used to extract extra data from incoming 172 message dicts. This version will work for both handle_raw_message() 173 and handle_raw_message_async() 174 """ 175 assert self._decode_filter_call is None 176 self._decode_filter_call = call 177 return call
Function decorator for defining a decode filter.
Decode filters can be used to extract extra data from incoming message dicts. This version will work for both handle_raw_message() and handle_raw_message_async()
def
encode_filter_method( self, call: Callable[[Any, Message | None, Response | SysResponse, dict], NoneType]) -> Callable[[Any, Message | None, Response, dict], NoneType]:
179 def encode_filter_method( 180 self, 181 call: Callable[ 182 [Any, Message | None, Response | SysResponse, dict], None 183 ], 184 ) -> Callable[[Any, Message | None, Response, dict], None]: 185 """Function decorator for defining an encode filter. 186 187 Encode filters can be used to add extra data to the message 188 dict before is is encoded to a string and sent out. 189 """ 190 assert self._encode_filter_call is None 191 self._encode_filter_call = call 192 return call
Function decorator for defining an encode filter.
Encode filters can be used to add extra data to the message dict before is is encoded to a string and sent out.
def
validate(self, log_only: bool = False) -> None:
194 def validate(self, log_only: bool = False) -> None: 195 """Check for handler completeness, valid types, etc.""" 196 for msgtype in self.protocol.message_ids_by_type.keys(): 197 if issubclass(msgtype, Response): 198 continue 199 if msgtype not in self._handlers: 200 msg = ( 201 f'Protocol message type {msgtype} is not handled' 202 f' by receiver type {type(self)}.' 203 ) 204 if log_only: 205 logging.error(msg) 206 else: 207 raise TypeError(msg)
Check for handler completeness, valid types, etc.
def
encode_user_response( self, bound_obj: Any, message: Message, response: Response | None) -> str:
226 def encode_user_response( 227 self, bound_obj: Any, message: Message, response: Response | None 228 ) -> str: 229 """Encode a response provided by the user for sending.""" 230 231 assert isinstance(response, Response | None) 232 # (user should never explicitly return error-responses) 233 assert ( 234 response is None or type(response) in message.get_response_types() 235 ) 236 237 # A return value of None equals EmptySysResponse. 238 out_response: Response | SysResponse 239 if response is None: 240 out_response = EmptySysResponse() 241 else: 242 out_response = response 243 244 response_dict = self.protocol.response_to_dict(out_response) 245 if self._encode_filter_call is not None: 246 self._encode_filter_call( 247 bound_obj, message, out_response, response_dict 248 ) 249 return self.protocol.encode_dict(response_dict)
Encode a response provided by the user for sending.
def
encode_error_response( self, bound_obj: Any, message: Message | None, exc: Exception) -> tuple[str, bool]:
251 def encode_error_response( 252 self, bound_obj: Any, message: Message | None, exc: Exception 253 ) -> tuple[str, bool]: 254 """Given an error, return sysresponse str and whether to log.""" 255 response, dolog = self.protocol.error_to_response(exc) 256 response_dict = self.protocol.response_to_dict(response) 257 if self._encode_filter_call is not None: 258 self._encode_filter_call( 259 bound_obj, message, response, response_dict 260 ) 261 return self.protocol.encode_dict(response_dict), dolog
Given an error, return sysresponse str and whether to log.
def
handle_raw_message(self, bound_obj: Any, msg: str, raise_unregistered: bool = False) -> str:
263 def handle_raw_message( 264 self, bound_obj: Any, msg: str, raise_unregistered: bool = False 265 ) -> str: 266 """Decode, handle, and return an response for a message. 267 268 if 'raise_unregistered' is True, will raise an 269 efro.message.UnregisteredMessageIDError for messages not handled by 270 the protocol. In all other cases local errors will translate to 271 error responses returned to the sender. 272 """ 273 assert not self.is_async, "can't call sync handler on async receiver" 274 msg_decoded: Message | None = None 275 try: 276 msg_decoded = self._decode_incoming_message(bound_obj, msg) 277 msgtype = type(msg_decoded) 278 handler = self._handlers.get(msgtype) 279 if handler is None: 280 raise RuntimeError(f'Got unhandled message type: {msgtype}.') 281 response = handler(bound_obj, msg_decoded) 282 assert isinstance(response, Response | None) 283 return self.encode_user_response(bound_obj, msg_decoded, response) 284 285 except Exception as exc: 286 if raise_unregistered and isinstance( 287 exc, UnregisteredMessageIDError 288 ): 289 raise 290 rstr, dolog = self.encode_error_response( 291 bound_obj, msg_decoded, exc 292 ) 293 if dolog: 294 if msg_decoded is not None: 295 msgtype = type(msg_decoded) 296 logging.exception( 297 'Error handling %s.%s message.', 298 msgtype.__module__, 299 msgtype.__qualname__, 300 ) 301 else: 302 logging.exception( 303 'Error handling raw efro.message' 304 ' (likely a message format incompatibility): %s.', 305 msg, 306 ) 307 return rstr
Decode, handle, and return an response for a message.
if 'raise_unregistered' is True, will raise an UnregisteredMessageIDError for messages not handled by the protocol. In all other cases local errors will translate to error responses returned to the sender.
def
handle_raw_message_async( self, bound_obj: Any, msg: str, raise_unregistered: bool = False) -> Awaitable[str]:
309 def handle_raw_message_async( 310 self, bound_obj: Any, msg: str, raise_unregistered: bool = False 311 ) -> Awaitable[str]: 312 """Should be called when the receiver gets a message. 313 314 The return value is the raw response to the message. 315 """ 316 317 # Note: This call is synchronous so that the first part of it can 318 # happen synchronously. If the whole call were async we wouldn't be 319 # able to guarantee that messages handlers would be called in the 320 # order the messages were received. 321 322 assert self.is_async, "Can't call async handler on sync receiver." 323 msg_decoded: Message | None = None 324 try: 325 msg_decoded = self._decode_incoming_message(bound_obj, msg) 326 msgtype = type(msg_decoded) 327 handler = self._handlers.get(msgtype) 328 if handler is None: 329 raise RuntimeError(f'Got unhandled message type: {msgtype}.') 330 handler_awaitable = handler(bound_obj, msg_decoded) 331 332 except Exception as exc: 333 if raise_unregistered and isinstance( 334 exc, UnregisteredMessageIDError 335 ): 336 raise 337 return self._handle_raw_message_async_error( 338 bound_obj, msg, msg_decoded, exc 339 ) 340 341 # Return an awaitable to handle the rest asynchronously. 342 return self._handle_raw_message_async( 343 bound_obj, msg, msg_decoded, handler_awaitable 344 )
Should be called when the receiver gets a message.
The return value is the raw response to the message.
class
BoundMessageReceiver:
400class BoundMessageReceiver: 401 """Base bound receiver class.""" 402 403 def __init__( 404 self, 405 obj: Any, 406 receiver: MessageReceiver, 407 ) -> None: 408 assert obj is not None 409 self._obj = obj 410 self._receiver = receiver 411 412 @property 413 def protocol(self) -> MessageProtocol: 414 """Protocol associated with this receiver.""" 415 return self._receiver.protocol 416 417 def encode_error_response(self, exc: Exception) -> str: 418 """Given an error, return a response ready to send. 419 420 This should be used for any errors that happen outside of 421 standard handle_raw_message calls. Any errors within those 422 calls will be automatically returned as encoded strings. 423 """ 424 # Passing None for Message here; we would only have that available 425 # for things going wrong in the handler (which this is not for). 426 return self._receiver.encode_error_response(self._obj, None, exc)[0]
Base bound receiver class.
BoundMessageReceiver(obj: Any, receiver: MessageReceiver)
protocol: MessageProtocol
412 @property 413 def protocol(self) -> MessageProtocol: 414 """Protocol associated with this receiver.""" 415 return self._receiver.protocol
Protocol associated with this receiver.
def
encode_error_response(self, exc: Exception) -> str:
417 def encode_error_response(self, exc: Exception) -> str: 418 """Given an error, return a response ready to send. 419 420 This should be used for any errors that happen outside of 421 standard handle_raw_message calls. Any errors within those 422 calls will be automatically returned as encoded strings. 423 """ 424 # Passing None for Message here; we would only have that available 425 # for things going wrong in the handler (which this is not for). 426 return self._receiver.encode_error_response(self._obj, None, exc)[0]
Given an error, return a response ready to send.
This should be used for any errors that happen outside of standard handle_raw_message calls. Any errors within those calls will be automatically returned as encoded strings.
def
create_sender_module( basename: str, protocol_create_code: str, enable_sync_sends: bool, enable_async_sends: bool, private: bool = False, protocol_module_level_import_code: str | None = None, build_time_protocol_create_code: str | None = None) -> str:
18def create_sender_module( 19 basename: str, 20 protocol_create_code: str, 21 enable_sync_sends: bool, 22 enable_async_sends: bool, 23 private: bool = False, 24 protocol_module_level_import_code: str | None = None, 25 build_time_protocol_create_code: str | None = None, 26) -> str: 27 """Create a Python module defining a MessageSender subclass. 28 29 This class is primarily for type checking and will contain overrides 30 for the varieties of send calls for message/response types defined 31 in the protocol. 32 33 Code passed for 'protocol_create_code' should import necessary 34 modules and assign an instance of the Protocol to a 'protocol' 35 variable. 36 37 Class names are based on basename; a basename 'FooSender' will 38 result in classes FooSender and BoundFooSender. 39 40 If 'private' is True, class-names will be prefixed with an '_'. 41 42 Note: output code may have long lines and should generally be run 43 through a formatter. We should perhaps move this functionality to 44 efrotools so we can include that functionality inline. 45 """ 46 protocol = _protocol_from_code( 47 build_time_protocol_create_code 48 if build_time_protocol_create_code is not None 49 else protocol_create_code 50 ) 51 return protocol.do_create_sender_module( 52 basename=basename, 53 protocol_create_code=protocol_create_code, 54 enable_sync_sends=enable_sync_sends, 55 enable_async_sends=enable_async_sends, 56 private=private, 57 protocol_module_level_import_code=protocol_module_level_import_code, 58 )
Create a Python module defining a MessageSender subclass.
This class is primarily for type checking and will contain overrides for the varieties of send calls for message/response types defined in the protocol.
Code passed for 'protocol_create_code' should import necessary modules and assign an instance of the Protocol to a 'protocol' variable.
Class names are based on basename; a basename 'FooSender' will result in classes FooSender and BoundFooSender.
If 'private' is True, class-names will be prefixed with an '_'.
Note: output code may have long lines and should generally be run through a formatter. We should perhaps move this functionality to efrotools so we can include that functionality inline.
def
create_receiver_module( basename: str, protocol_create_code: str, is_async: bool, private: bool = False, protocol_module_level_import_code: str | None = None, build_time_protocol_create_code: str | None = None) -> str:
61def create_receiver_module( 62 basename: str, 63 protocol_create_code: str, 64 is_async: bool, 65 private: bool = False, 66 protocol_module_level_import_code: str | None = None, 67 build_time_protocol_create_code: str | None = None, 68) -> str: 69 """ "Create a Python module defining a MessageReceiver subclass. 70 71 This class is primarily for type checking and will contain overrides 72 for the register method for message/response types defined in 73 the protocol. 74 75 Class names are based on basename; a basename 'FooReceiver' will 76 result in FooReceiver and BoundFooReceiver. 77 78 If 'is_async' is True, handle_raw_message() will be an async method 79 and the @handler decorator will expect async methods. 80 81 If 'private' is True, class-names will be prefixed with an '_'. 82 83 Note that line lengths are not clipped, so output may need to be 84 run through a formatter to prevent lint warnings about excessive 85 line lengths. 86 """ 87 protocol = _protocol_from_code( 88 build_time_protocol_create_code 89 if build_time_protocol_create_code is not None 90 else protocol_create_code 91 ) 92 return protocol.do_create_receiver_module( 93 basename=basename, 94 protocol_create_code=protocol_create_code, 95 is_async=is_async, 96 private=private, 97 protocol_module_level_import_code=protocol_module_level_import_code, 98 )
"Create a Python module defining a MessageReceiver subclass.
This class is primarily for type checking and will contain overrides for the register method for message/response types defined in the protocol.
Class names are based on basename; a basename 'FooReceiver' will result in FooReceiver and BoundFooReceiver.
If 'is_async' is True, handle_raw_message() will be an async method and the @handler decorator will expect async methods.
If 'private' is True, class-names will be prefixed with an '_'.
Note that line lengths are not clipped, so output may need to be run through a formatter to prevent lint warnings about excessive line lengths.
class
UnregisteredMessageIDError(builtins.Exception):
20class UnregisteredMessageIDError(Exception): 21 """A message or response id is not covered by our protocol."""
A message or response id is not covered by our protocol.
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- add_note
- args