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.

@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
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:
36class Response:
37    """Base class for responses to messages."""

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.

def get_local_exception(self) -> Exception | None:
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

Fetch a local attached exception.

@ioprepped
@dataclass
class EmptySysResponse(efro.message.SysResponse):
86@ioprepped
87@dataclass
88class EmptySysResponse(SysResponse):
89    """The response equivalent of None."""

The response equivalent of None.

@ioprepped
@dataclass
class ErrorSysResponse(efro.message.SysResponse):
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._base.IOAttrs object>], error_type: Annotated[ErrorSysResponse.ErrorType, <efro.dataclassio._base.IOAttrs object>] = <ErrorType.REMOTE: 0>)
error_message: Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x12aef5760>]
error_type: Annotated[ErrorSysResponse.ErrorType, <efro.dataclassio._base.IOAttrs object at 0x12aef5a60>] = <ErrorType.REMOTE: 0>
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.

REMOTE = <ErrorType.REMOTE: 0>
REMOTE_CLEAN = <ErrorType.REMOTE_CLEAN: 1>
LOCAL = <ErrorType.LOCAL: 2>
COMMUNICATION = <ErrorType.COMMUNICATION: 3>
REMOTE_COMMUNICATION = <ErrorType.REMOTE_COMMUNICATION: 4>
Inherited Members
enum.Enum
name
value
@ioprepped
@dataclass
class StringResponse(efro.message.Response):
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._base.IOAttrs object>])
value: Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x12aef7680>]
@ioprepped
@dataclass
class BoolResponse(efro.message.Response):
 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._base.IOAttrs object>])
value: Annotated[bool, <efro.dataclassio._base.IOAttrs object at 0x12aefe600>]
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]
forward_clean_errors
forward_communication_errors
remote_errors_include_stack_traces
log_errors_on_receiver
@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.

def message_to_dict(self, message: Message) -> 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.

def response_to_dict( self, response: Response | SysResponse) -> 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.

def error_to_response(self, exc: Exception) -> tuple[SysResponse, bool]:
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.

def message_from_dict(self, data: dict) -> Message:
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.

def response_from_dict( self, data: dict) -> Response | SysResponse:
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_async_raw_message_call: (
 57            Callable[[Any, str], Awaitable[str]] | None
 58        ) = None
 59        self._send_async_raw_message_ex_call: (
 60            Callable[[Any, str, Message], Awaitable[str]] | None
 61        ) = None
 62        self._encode_filter_call: (
 63            Callable[[Any, Message, dict], None] | None
 64        ) = None
 65        self._decode_filter_call: (
 66            Callable[[Any, Message, dict, Response | SysResponse], None] | None
 67        ) = None
 68        self._peer_desc_call: Callable[[Any], str] | None = None
 69
 70    def send_method(
 71        self, call: Callable[[Any, str], str]
 72    ) -> Callable[[Any, str], str]:
 73        """Function decorator for setting raw send method.
 74
 75        Send methods take strings and should return strings.
 76        CommunicationErrors raised here will be returned to the sender
 77        as such; all other exceptions will result in a RuntimeError for
 78        the sender.
 79        """
 80        assert self._send_raw_message_call is None
 81        self._send_raw_message_call = call
 82        return call
 83
 84    def send_async_method(
 85        self, call: Callable[[Any, str], Awaitable[str]]
 86    ) -> Callable[[Any, str], Awaitable[str]]:
 87        """Function decorator for setting raw send-async method.
 88
 89        Send methods take strings and should return strings.
 90        CommunicationErrors raised here will be returned to the sender
 91        as such; all other exceptions will result in a RuntimeError for
 92        the sender.
 93
 94        IMPORTANT: Generally async send methods should not be implemented
 95        as 'async' methods, but instead should be regular methods that
 96        return awaitable objects. This way it can be guaranteed that
 97        outgoing messages are synchronously enqueued in the correct
 98        order, and then async calls can be returned which finish each
 99        send. If the entire call is async, they may be enqueued out of
100        order in rare cases.
101        """
102        assert self._send_async_raw_message_call is None
103        self._send_async_raw_message_call = call
104        return call
105
106    def send_async_ex_method(
107        self, call: Callable[[Any, str, Message], Awaitable[str]]
108    ) -> Callable[[Any, str, Message], Awaitable[str]]:
109        """Function decorator for extended send-async method.
110
111        Version of send_async_method which is also is passed the original
112        unencoded message; can be useful for cases where metadata is sent
113        along with messages referring to their payloads/etc.
114        """
115        assert self._send_async_raw_message_ex_call is None
116        self._send_async_raw_message_ex_call = call
117        return call
118
119    def encode_filter_method(
120        self, call: Callable[[Any, Message, dict], None]
121    ) -> Callable[[Any, Message, dict], None]:
122        """Function decorator for defining an encode filter.
123
124        Encode filters can be used to add extra data to the message
125        dict before is is encoded to a string and sent out.
126        """
127        assert self._encode_filter_call is None
128        self._encode_filter_call = call
129        return call
130
131    def decode_filter_method(
132        self, call: Callable[[Any, Message, dict, Response | SysResponse], None]
133    ) -> Callable[[Any, Message, dict, Response], None]:
134        """Function decorator for defining a decode filter.
135
136        Decode filters can be used to extract extra data from incoming
137        message dicts.
138        """
139        assert self._decode_filter_call is None
140        self._decode_filter_call = call
141        return call
142
143    def peer_desc_method(
144        self, call: Callable[[Any], str]
145    ) -> Callable[[Any], str]:
146        """Function decorator for defining peer descriptions.
147
148        These are included in error messages or other diagnostics.
149        """
150        assert self._peer_desc_call is None
151        self._peer_desc_call = call
152        return call
153
154    def send(self, bound_obj: Any, message: Message) -> Response | None:
155        """Send a message synchronously."""
156        return self.unpack_raw_response(
157            bound_obj=bound_obj,
158            message=message,
159            raw_response=self.fetch_raw_response(
160                bound_obj=bound_obj,
161                message=message,
162            ),
163        )
164
165    def send_async(
166        self, bound_obj: Any, message: Message
167    ) -> Awaitable[Response | None]:
168        """Send a message asynchronously."""
169
170        # Note: This call is synchronous so that the first part of it can
171        # happen synchronously. If the whole call were async we wouldn't be
172        # able to guarantee that messages sent in order would actually go
173        # out in order.
174        raw_response_awaitable = self.fetch_raw_response_async(
175            bound_obj=bound_obj,
176            message=message,
177        )
178        # Now return an awaitable that will finish the send.
179        return self._send_async_awaitable(
180            bound_obj, message, raw_response_awaitable
181        )
182
183    async def _send_async_awaitable(
184        self,
185        bound_obj: Any,
186        message: Message,
187        raw_response_awaitable: Awaitable[Response | SysResponse],
188    ) -> Response | None:
189        return self.unpack_raw_response(
190            bound_obj=bound_obj,
191            message=message,
192            raw_response=await raw_response_awaitable,
193        )
194
195    def fetch_raw_response(
196        self, bound_obj: Any, message: Message
197    ) -> Response | SysResponse:
198        """Send a message synchronously.
199
200        Generally you can just call send(); these split versions are
201        for when message sending and response handling need to happen
202        in different contexts/threads.
203        """
204        if self._send_raw_message_call is None:
205            raise RuntimeError('send() is unimplemented for this type.')
206
207        msg_encoded = self._encode_message(bound_obj, message)
208        try:
209            response_encoded = self._send_raw_message_call(
210                bound_obj, msg_encoded
211            )
212        except Exception as exc:
213            response = ErrorSysResponse(
214                error_message='Error in MessageSender @send_method.',
215                error_type=(
216                    ErrorSysResponse.ErrorType.COMMUNICATION
217                    if isinstance(exc, CommunicationError)
218                    else ErrorSysResponse.ErrorType.LOCAL
219                ),
220            )
221            # Can include the actual exception since we'll be looking at
222            # this locally; might be helpful.
223            response.set_local_exception(exc)
224            return response
225        return self._decode_raw_response(bound_obj, message, response_encoded)
226
227    def fetch_raw_response_async(
228        self, bound_obj: Any, message: Message
229    ) -> Awaitable[Response | SysResponse]:
230        """Fetch a raw message response awaitable.
231
232        The result of this should be awaited and then passed to
233        unpack_raw_response() to produce the final message result.
234
235        Generally you can just call send(); calling fetch and unpack
236        manually is for when message sending and response handling need
237        to happen in different contexts/threads.
238        """
239
240        # Note: This call is synchronous so that the first part of it can
241        # happen synchronously. If the whole call were async we wouldn't be
242        # able to guarantee that messages sent in order would actually go
243        # out in order.
244        if (
245            self._send_async_raw_message_call is None
246            and self._send_async_raw_message_ex_call is None
247        ):
248            raise RuntimeError('send_async() is unimplemented for this type.')
249
250        msg_encoded = self._encode_message(bound_obj, message)
251        try:
252            if self._send_async_raw_message_ex_call is not None:
253                send_awaitable = self._send_async_raw_message_ex_call(
254                    bound_obj, msg_encoded, message
255                )
256            else:
257                assert self._send_async_raw_message_call is not None
258                send_awaitable = self._send_async_raw_message_call(
259                    bound_obj, msg_encoded
260                )
261        except Exception as exc:
262            return self._error_awaitable(exc)
263
264        # Now return an awaitable to finish the job.
265        return self._fetch_raw_response_awaitable(
266            bound_obj, message, send_awaitable
267        )
268
269    async def _error_awaitable(self, exc: Exception) -> SysResponse:
270        response = ErrorSysResponse(
271            error_message='Error in MessageSender @send_async_method.',
272            error_type=(
273                ErrorSysResponse.ErrorType.COMMUNICATION
274                if isinstance(exc, CommunicationError)
275                else ErrorSysResponse.ErrorType.LOCAL
276            ),
277        )
278        # Can include the actual exception since we'll be looking at
279        # this locally; might be helpful.
280        response.set_local_exception(exc)
281        return response
282
283    async def _fetch_raw_response_awaitable(
284        self, bound_obj: Any, message: Message, send_awaitable: Awaitable[str]
285    ) -> Response | SysResponse:
286        try:
287            response_encoded = await send_awaitable
288        except Exception as exc:
289            response = ErrorSysResponse(
290                error_message='Error in MessageSender @send_async_method.',
291                error_type=(
292                    ErrorSysResponse.ErrorType.COMMUNICATION
293                    if isinstance(exc, CommunicationError)
294                    else ErrorSysResponse.ErrorType.LOCAL
295                ),
296            )
297            # Can include the actual exception since we'll be looking at
298            # this locally; might be helpful.
299            response.set_local_exception(exc)
300            return response
301        return self._decode_raw_response(bound_obj, message, response_encoded)
302
303    def unpack_raw_response(
304        self,
305        bound_obj: Any,
306        message: Message,
307        raw_response: Response | SysResponse,
308    ) -> Response | None:
309        """Convert a raw fetched response into a final response/error/etc.
310
311        Generally you can just call send(); calling fetch and unpack
312        manually is for when message sending and response handling need
313        to happen in different contexts/threads.
314        """
315        response = self._unpack_raw_response(bound_obj, raw_response)
316        assert (
317            response is None
318            or type(response) in type(message).get_response_types()
319        )
320        return response
321
322    def _encode_message(self, bound_obj: Any, message: Message) -> str:
323        """Encode a message for sending."""
324        msg_dict = self.protocol.message_to_dict(message)
325        if self._encode_filter_call is not None:
326            self._encode_filter_call(bound_obj, message, msg_dict)
327        return self.protocol.encode_dict(msg_dict)
328
329    def _decode_raw_response(
330        self, bound_obj: Any, message: Message, response_encoded: str
331    ) -> Response | SysResponse:
332        """Create a Response from returned data.
333
334        These Responses may encapsulate things like remote errors and
335        should not be handed directly to users. _unpack_raw_response()
336        should be used to translate to special values like None or raise
337        Exceptions. This function itself should never raise Exceptions.
338        """
339        response: Response | SysResponse
340        try:
341            response_dict = self.protocol.decode_dict(response_encoded)
342            response = self.protocol.response_from_dict(response_dict)
343            if self._decode_filter_call is not None:
344                self._decode_filter_call(
345                    bound_obj, message, response_dict, response
346                )
347        except Exception as exc:
348            response = ErrorSysResponse(
349                error_message='Error decoding raw response.',
350                error_type=ErrorSysResponse.ErrorType.LOCAL,
351            )
352            # Since we'll be looking at this locally, we can include
353            # extra info for logging/etc.
354            response.set_local_exception(exc)
355        return response
356
357    def _unpack_raw_response(
358        self, bound_obj: Any, raw_response: Response | SysResponse
359    ) -> Response | None:
360        """Given a raw Response, unpacks to special values or Exceptions.
361
362        The result of this call is what should be passed to users.
363        For complex messaging situations such as response callbacks
364        operating across different threads, this last stage should be
365        run such that any raised Exception is active when the callback
366        fires; not on the thread where the message was sent.
367        """
368        # EmptySysResponse translates to None
369        if isinstance(raw_response, EmptySysResponse):
370            return None
371
372        # Some error occurred. Raise a local Exception for it.
373        if isinstance(raw_response, ErrorSysResponse):
374            # Errors that happened locally can attach their exceptions
375            # here for extra logging goodness.
376            local_exception = raw_response.get_local_exception()
377
378            if (
379                raw_response.error_type
380                is ErrorSysResponse.ErrorType.COMMUNICATION
381            ):
382                raise CommunicationError(
383                    raw_response.error_message
384                ) from local_exception
385
386            # If something went wrong on *our* end of the connection,
387            # don't say it was a remote error.
388            if raw_response.error_type is ErrorSysResponse.ErrorType.LOCAL:
389                raise RuntimeError(
390                    raw_response.error_message
391                ) from local_exception
392
393            # If they want to support clean errors, do those.
394            if (
395                self.protocol.forward_clean_errors
396                and raw_response.error_type
397                is ErrorSysResponse.ErrorType.REMOTE_CLEAN
398            ):
399                raise CleanError(
400                    raw_response.error_message
401                ) from local_exception
402
403            if (
404                self.protocol.forward_communication_errors
405                and raw_response.error_type
406                is ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION
407            ):
408                raise CommunicationError(
409                    raw_response.error_message
410                ) from local_exception
411
412            # Everything else gets lumped in as a remote error.
413            raise RemoteError(
414                raw_response.error_message,
415                peer_desc=(
416                    'peer'
417                    if self._peer_desc_call is None
418                    else self._peer_desc_call(bound_obj)
419                ),
420            ) from local_exception
421
422        assert isinstance(raw_response, Response)
423        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_async_raw_message_call: (
57            Callable[[Any, str], Awaitable[str]] | None
58        ) = None
59        self._send_async_raw_message_ex_call: (
60            Callable[[Any, str, Message], Awaitable[str]] | None
61        ) = None
62        self._encode_filter_call: (
63            Callable[[Any, Message, dict], None] | None
64        ) = None
65        self._decode_filter_call: (
66            Callable[[Any, Message, dict, Response | SysResponse], None] | None
67        ) = None
68        self._peer_desc_call: Callable[[Any], str] | None = None
protocol
def send_method(self, call: Callable[[Any, str], str]) -> Callable[[Any, str], str]:
70    def send_method(
71        self, call: Callable[[Any, str], str]
72    ) -> Callable[[Any, str], str]:
73        """Function decorator for setting raw send method.
74
75        Send methods take strings and should return strings.
76        CommunicationErrors raised here will be returned to the sender
77        as such; all other exceptions will result in a RuntimeError for
78        the sender.
79        """
80        assert self._send_raw_message_call is None
81        self._send_raw_message_call = call
82        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_async_method( self, call: Callable[[Any, str], Awaitable[str]]) -> Callable[[Any, str], Awaitable[str]]:
 84    def send_async_method(
 85        self, call: Callable[[Any, str], Awaitable[str]]
 86    ) -> Callable[[Any, str], Awaitable[str]]:
 87        """Function decorator for setting raw send-async method.
 88
 89        Send methods take strings and should return strings.
 90        CommunicationErrors raised here will be returned to the sender
 91        as such; all other exceptions will result in a RuntimeError for
 92        the sender.
 93
 94        IMPORTANT: Generally async send methods should not be implemented
 95        as 'async' methods, but instead should be regular methods that
 96        return awaitable objects. This way it can be guaranteed that
 97        outgoing messages are synchronously enqueued in the correct
 98        order, and then async calls can be returned which finish each
 99        send. If the entire call is async, they may be enqueued out of
100        order in rare cases.
101        """
102        assert self._send_async_raw_message_call is None
103        self._send_async_raw_message_call = call
104        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]]:
106    def send_async_ex_method(
107        self, call: Callable[[Any, str, Message], Awaitable[str]]
108    ) -> Callable[[Any, str, Message], Awaitable[str]]:
109        """Function decorator for extended send-async method.
110
111        Version of send_async_method which is also is passed the original
112        unencoded message; can be useful for cases where metadata is sent
113        along with messages referring to their payloads/etc.
114        """
115        assert self._send_async_raw_message_ex_call is None
116        self._send_async_raw_message_ex_call = call
117        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]:
119    def encode_filter_method(
120        self, call: Callable[[Any, Message, dict], None]
121    ) -> Callable[[Any, Message, dict], None]:
122        """Function decorator for defining an encode filter.
123
124        Encode filters can be used to add extra data to the message
125        dict before is is encoded to a string and sent out.
126        """
127        assert self._encode_filter_call is None
128        self._encode_filter_call = call
129        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]:
131    def decode_filter_method(
132        self, call: Callable[[Any, Message, dict, Response | SysResponse], None]
133    ) -> Callable[[Any, Message, dict, Response], None]:
134        """Function decorator for defining a decode filter.
135
136        Decode filters can be used to extract extra data from incoming
137        message dicts.
138        """
139        assert self._decode_filter_call is None
140        self._decode_filter_call = call
141        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]:
143    def peer_desc_method(
144        self, call: Callable[[Any], str]
145    ) -> Callable[[Any], str]:
146        """Function decorator for defining peer descriptions.
147
148        These are included in error messages or other diagnostics.
149        """
150        assert self._peer_desc_call is None
151        self._peer_desc_call = call
152        return call

Function decorator for defining peer descriptions.

These are included in error messages or other diagnostics.

def send( self, bound_obj: Any, message: Message) -> Response | None:
154    def send(self, bound_obj: Any, message: Message) -> Response | None:
155        """Send a message synchronously."""
156        return self.unpack_raw_response(
157            bound_obj=bound_obj,
158            message=message,
159            raw_response=self.fetch_raw_response(
160                bound_obj=bound_obj,
161                message=message,
162            ),
163        )

Send a message synchronously.

def send_async( self, bound_obj: Any, message: Message) -> Awaitable[Response | None]:
165    def send_async(
166        self, bound_obj: Any, message: Message
167    ) -> Awaitable[Response | None]:
168        """Send a message asynchronously."""
169
170        # Note: This call is synchronous so that the first part of it can
171        # happen synchronously. If the whole call were async we wouldn't be
172        # able to guarantee that messages sent in order would actually go
173        # out in order.
174        raw_response_awaitable = self.fetch_raw_response_async(
175            bound_obj=bound_obj,
176            message=message,
177        )
178        # Now return an awaitable that will finish the send.
179        return self._send_async_awaitable(
180            bound_obj, message, raw_response_awaitable
181        )

Send a message asynchronously.

def fetch_raw_response( self, bound_obj: Any, message: Message) -> Response | SysResponse:
195    def fetch_raw_response(
196        self, bound_obj: Any, message: Message
197    ) -> Response | SysResponse:
198        """Send a message synchronously.
199
200        Generally you can just call send(); these split versions are
201        for when message sending and response handling need to happen
202        in different contexts/threads.
203        """
204        if self._send_raw_message_call is None:
205            raise RuntimeError('send() is unimplemented for this type.')
206
207        msg_encoded = self._encode_message(bound_obj, message)
208        try:
209            response_encoded = self._send_raw_message_call(
210                bound_obj, msg_encoded
211            )
212        except Exception as exc:
213            response = ErrorSysResponse(
214                error_message='Error in MessageSender @send_method.',
215                error_type=(
216                    ErrorSysResponse.ErrorType.COMMUNICATION
217                    if isinstance(exc, CommunicationError)
218                    else ErrorSysResponse.ErrorType.LOCAL
219                ),
220            )
221            # Can include the actual exception since we'll be looking at
222            # this locally; might be helpful.
223            response.set_local_exception(exc)
224            return response
225        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]:
227    def fetch_raw_response_async(
228        self, bound_obj: Any, message: Message
229    ) -> Awaitable[Response | SysResponse]:
230        """Fetch a raw message response awaitable.
231
232        The result of this should be awaited and then passed to
233        unpack_raw_response() to produce the final message result.
234
235        Generally you can just call send(); calling fetch and unpack
236        manually is for when message sending and response handling need
237        to happen in different contexts/threads.
238        """
239
240        # Note: This call is synchronous so that the first part of it can
241        # happen synchronously. If the whole call were async we wouldn't be
242        # able to guarantee that messages sent in order would actually go
243        # out in order.
244        if (
245            self._send_async_raw_message_call is None
246            and self._send_async_raw_message_ex_call is None
247        ):
248            raise RuntimeError('send_async() is unimplemented for this type.')
249
250        msg_encoded = self._encode_message(bound_obj, message)
251        try:
252            if self._send_async_raw_message_ex_call is not None:
253                send_awaitable = self._send_async_raw_message_ex_call(
254                    bound_obj, msg_encoded, message
255                )
256            else:
257                assert self._send_async_raw_message_call is not None
258                send_awaitable = self._send_async_raw_message_call(
259                    bound_obj, msg_encoded
260                )
261        except Exception as exc:
262            return self._error_awaitable(exc)
263
264        # Now return an awaitable to finish the job.
265        return self._fetch_raw_response_awaitable(
266            bound_obj, message, send_awaitable
267        )

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:
303    def unpack_raw_response(
304        self,
305        bound_obj: Any,
306        message: Message,
307        raw_response: Response | SysResponse,
308    ) -> Response | None:
309        """Convert a raw fetched response into a final response/error/etc.
310
311        Generally you can just call send(); calling fetch and unpack
312        manually is for when message sending and response handling need
313        to happen in different contexts/threads.
314        """
315        response = self._unpack_raw_response(bound_obj, raw_response)
316        assert (
317            response is None
318            or type(response) in type(message).get_response_types()
319        )
320        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:
426class BoundMessageSender:
427    """Base class for bound senders."""
428
429    def __init__(self, obj: Any, sender: MessageSender) -> None:
430        # Note: not checking obj here since we want to support
431        # at least our protocol property when accessed via type.
432        self._obj = obj
433        self._sender = sender
434
435    @property
436    def protocol(self) -> MessageProtocol:
437        """Protocol associated with this sender."""
438        return self._sender.protocol
439
440    def send_untyped(self, message: Message) -> Response | None:
441        """Send a message synchronously.
442
443        Whenever possible, use the send() call provided by generated
444        subclasses instead of this; it will provide better type safety.
445        """
446        assert self._obj is not None
447        return self._sender.send(bound_obj=self._obj, message=message)
448
449    def send_async_untyped(
450        self, message: Message
451    ) -> Awaitable[Response | None]:
452        """Send a message asynchronously.
453
454        Whenever possible, use the send_async() call provided by generated
455        subclasses instead of this; it will provide better type safety.
456        """
457        assert self._obj is not None
458        return self._sender.send_async(bound_obj=self._obj, message=message)
459
460    def fetch_raw_response_async_untyped(
461        self, message: Message
462    ) -> Awaitable[Response | SysResponse]:
463        """Split send (part 1 of 2)."""
464        assert self._obj is not None
465        return self._sender.fetch_raw_response_async(
466            bound_obj=self._obj, message=message
467        )
468
469    def unpack_raw_response_untyped(
470        self, message: Message, raw_response: Response | SysResponse
471    ) -> Response | None:
472        """Split send (part 2 of 2)."""
473        return self._sender.unpack_raw_response(
474            bound_obj=self._obj, message=message, raw_response=raw_response
475        )

Base class for bound senders.

BoundMessageSender(obj: Any, sender: MessageSender)
429    def __init__(self, obj: Any, sender: MessageSender) -> None:
430        # Note: not checking obj here since we want to support
431        # at least our protocol property when accessed via type.
432        self._obj = obj
433        self._sender = sender
protocol: MessageProtocol
435    @property
436    def protocol(self) -> MessageProtocol:
437        """Protocol associated with this sender."""
438        return self._sender.protocol

Protocol associated with this sender.

def send_untyped( self, message: Message) -> Response | None:
440    def send_untyped(self, message: Message) -> Response | None:
441        """Send a message synchronously.
442
443        Whenever possible, use the send() call provided by generated
444        subclasses instead of this; it will provide better type safety.
445        """
446        assert self._obj is not None
447        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.

def send_async_untyped( self, message: Message) -> Awaitable[Response | None]:
449    def send_async_untyped(
450        self, message: Message
451    ) -> Awaitable[Response | None]:
452        """Send a message asynchronously.
453
454        Whenever possible, use the send_async() call provided by generated
455        subclasses instead of this; it will provide better type safety.
456        """
457        assert self._obj is not None
458        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.

def fetch_raw_response_async_untyped( self, message: Message) -> Awaitable[Response | SysResponse]:
460    def fetch_raw_response_async_untyped(
461        self, message: Message
462    ) -> Awaitable[Response | SysResponse]:
463        """Split send (part 1 of 2)."""
464        assert self._obj is not None
465        return self._sender.fetch_raw_response_async(
466            bound_obj=self._obj, message=message
467        )

Split send (part 1 of 2).

def unpack_raw_response_untyped( self, message: Message, raw_response: Response | SysResponse) -> Response | None:
469    def unpack_raw_response_untyped(
470        self, message: Message, raw_response: Response | SysResponse
471    ) -> Response | None:
472        """Split send (part 2 of 2)."""
473        return self._sender.unpack_raw_response(
474            bound_obj=self._obj, message=message, raw_response=raw_response
475        )

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
is_async = False
protocol
def register_handler( self, call: Callable[[Any, Message], Response | None]) -> 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 efro.message.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)
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
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