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: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x114bceb50>], error_type: typing.Annotated[ErrorSysResponse.ErrorType, <efro.dataclassio._base.IOAttrs object at 0x114bcea50>] = <ErrorType.REMOTE: 0>)
error_message: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x10c3be2d0>]
error_type: typing.Annotated[ErrorSysResponse.ErrorType, <efro.dataclassio._base.IOAttrs object at 0x114b3cc50>] = <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: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x114bc6850>])
value: typing.Annotated[str, <efro.dataclassio._base.IOAttrs object at 0x110bf5010>]
@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: typing.Annotated[bool, <efro.dataclassio._base.IOAttrs object at 0x114b77250>])
value: typing.Annotated[bool, <efro.dataclassio._base.IOAttrs object at 0x102dc08d0>]
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_remote_errors: 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
 67        end - at least 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_remote_errors' 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_remote_errors = log_remote_errors
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                    traceback.format_exc()
218                    if self.remote_errors_include_stack_traces
219                    else 'An internal error has occurred.'
220                ),
221                error_type=ErrorSysResponse.ErrorType.REMOTE,
222            ),
223            self.log_remote_errors,
224        )
225
226    def _to_dict(
227        self, message: Any, ids_by_type: dict[type, int], opname: str
228    ) -> dict:
229        """Encode a message to a json string for transport."""
230
231        m_id: int | None = ids_by_type.get(type(message))
232        if m_id is None:
233            raise TypeError(
234                f'{opname} type is not registered in protocol:'
235                f' {type(message)}'
236            )
237        out = {'t': m_id, 'm': dataclass_to_dict(message)}
238        return out
239
240    @staticmethod
241    def decode_dict(data: str) -> dict:
242        """Decode data to a dict."""
243        out = json.loads(data)
244        assert isinstance(out, dict)
245        return out
246
247    def message_from_dict(self, data: dict) -> Message:
248        """Decode a message from a json string."""
249        out = self._from_dict(data, self.message_types_by_id, 'message')
250        assert isinstance(out, Message)
251        return out
252
253    def response_from_dict(self, data: dict) -> Response | SysResponse:
254        """Decode a response from a json string."""
255        out = self._from_dict(data, self.response_types_by_id, 'response')
256        assert isinstance(out, Response | SysResponse)
257        return out
258
259    # Weeeird; we get mypy errors returning dict[int, type] but
260    # dict[int, typing.Type] or dict[int, type[Any]] works..
261    def _from_dict(
262        self, data: dict, types_by_id: dict[int, type[Any]], opname: str
263    ) -> Any:
264        """Decode a message from a json string."""
265        msgdict: dict | None
266
267        m_id = data.get('t')
268        # Allow omitting 'm' dict if its empty.
269        msgdict = data.get('m', {})
270
271        assert isinstance(m_id, int)
272        assert isinstance(msgdict, dict)
273
274        # Decode this particular type.
275        msgtype = types_by_id.get(m_id)
276        if msgtype is None:
277            raise UnregisteredMessageIDError(
278                f'Got unregistered {opname} id of {m_id}.'
279            )
280        return dataclass_from_dict(msgtype, msgdict)
281
282    def _get_module_header(
283        self,
284        part: Literal['sender', 'receiver'],
285        extra_import_code: str | None,
286        enable_async_sends: bool,
287    ) -> str:
288        """Return common parts of generated modules."""
289        # pylint: disable=too-many-locals
290        # pylint: disable=too-many-branches
291        # pylint: disable=too-many-statements
292        import textwrap
293
294        tpimports: dict[str, list[str]] = {}
295        imports: dict[str, list[str]] = {}
296
297        single_message_type = len(self.message_ids_by_type) == 1
298
299        msgtypes = list(self.message_ids_by_type)
300        if part == 'sender':
301            msgtypes.append(Message)
302        for msgtype in msgtypes:
303            tpimports.setdefault(msgtype.__module__, []).append(
304                msgtype.__name__
305            )
306        rsptypes = list(self.response_ids_by_type)
307        if part == 'sender':
308            rsptypes.append(Response)
309        for rsp_tp in rsptypes:
310            # Skip these as they don't actually show up in code.
311            if rsp_tp is EmptySysResponse or rsp_tp is ErrorSysResponse:
312                continue
313            if (
314                single_message_type
315                and part == 'sender'
316                and rsp_tp is not Response
317            ):
318                # We need to cast to the single supported response type
319                # in this case so need response types at runtime.
320                imports.setdefault(rsp_tp.__module__, []).append(
321                    rsp_tp.__name__
322                )
323            else:
324                tpimports.setdefault(rsp_tp.__module__, []).append(
325                    rsp_tp.__name__
326                )
327
328        import_lines = ''
329        tpimport_lines = ''
330
331        for module, names in sorted(imports.items()):
332            jnames = ', '.join(names)
333            line = f'from {module} import {jnames}'
334            if len(line) > 80:
335                # Recreate in a wrapping-friendly form.
336                line = f'from {module} import ({jnames})'
337            import_lines += f'{line}\n'
338        for module, names in sorted(tpimports.items()):
339            jnames = ', '.join(names)
340            line = f'from {module} import {jnames}'
341            if len(line) > 75:  # Account for indent
342                # Recreate in a wrapping-friendly form.
343                line = f'from {module} import ({jnames})'
344            tpimport_lines += f'{line}\n'
345
346        if part == 'sender':
347            import_lines += (
348                'from efro.message import MessageSender, BoundMessageSender'
349            )
350            tpimport_typing_extras = ''
351        else:
352            if single_message_type:
353                import_lines += (
354                    'from efro.message import (MessageReceiver,'
355                    ' BoundMessageReceiver, Message, Response)'
356                )
357            else:
358                import_lines += (
359                    'from efro.message import MessageReceiver,'
360                    ' BoundMessageReceiver'
361                )
362            tpimport_typing_extras = ', Awaitable'
363
364        if extra_import_code is not None:
365            import_lines += f'\n{extra_import_code}\n'
366
367        ovld = ', overload' if not single_message_type else ''
368        ovld2 = (
369            ', cast, Awaitable'
370            if (single_message_type and part == 'sender' and enable_async_sends)
371            else ''
372        )
373        tpimport_lines = textwrap.indent(tpimport_lines, '    ')
374
375        baseimps = ['Any']
376        if part == 'receiver':
377            baseimps.append('Callable')
378        if part == 'sender' and enable_async_sends:
379            baseimps.append('Awaitable')
380        baseimps_s = ', '.join(baseimps)
381        out = (
382            '# Released under the MIT License. See LICENSE for details.\n'
383            f'#\n'
384            f'"""Auto-generated {part} module. Do not edit by hand."""\n'
385            f'\n'
386            f'from __future__ import annotations\n'
387            f'\n'
388            f'from typing import TYPE_CHECKING{ovld}{ovld2}\n'
389            f'\n'
390            f'{import_lines}\n'
391            f'\n'
392            f'if TYPE_CHECKING:\n'
393            f'    from typing import {baseimps_s}'
394            f'{tpimport_typing_extras}\n'
395            f'{tpimport_lines}'
396            f'\n'
397            f'\n'
398        )
399        return out
400
401    def do_create_sender_module(
402        self,
403        basename: str,
404        protocol_create_code: str,
405        enable_sync_sends: bool,
406        enable_async_sends: bool,
407        private: bool = False,
408        protocol_module_level_import_code: str | None = None,
409    ) -> str:
410        """Used by create_sender_module(); do not call directly."""
411        # pylint: disable=too-many-locals
412        # pylint: disable=too-many-branches
413        import textwrap
414
415        msgtypes = list(self.message_ids_by_type.keys())
416
417        ppre = '_' if private else ''
418        out = self._get_module_header(
419            'sender',
420            extra_import_code=protocol_module_level_import_code,
421            enable_async_sends=enable_async_sends,
422        )
423        ccind = textwrap.indent(protocol_create_code, '        ')
424        out += (
425            f'class {ppre}{basename}(MessageSender):\n'
426            f'    """Protocol-specific sender."""\n'
427            f'\n'
428            f'    def __init__(self) -> None:\n'
429            f'{ccind}\n'
430            f'        super().__init__(protocol)\n'
431            f'\n'
432            f'    def __get__(\n'
433            f'        self, obj: Any, type_in: Any = None\n'
434            f'    ) -> {ppre}Bound{basename}:\n'
435            f'        return {ppre}Bound{basename}(obj, self)\n'
436            f'\n'
437            f'\n'
438            f'class {ppre}Bound{basename}(BoundMessageSender):\n'
439            f'    """Protocol-specific bound sender."""\n'
440        )
441
442        def _filt_tp_name(rtype: type[Response] | None) -> str:
443            return 'None' if rtype is None else rtype.__name__
444
445        # Define handler() overloads for all registered message types.
446        if msgtypes:
447            for async_pass in False, True:
448                if async_pass and not enable_async_sends:
449                    continue
450                if not async_pass and not enable_sync_sends:
451                    continue
452                pfx = 'async ' if async_pass else ''
453                sfx = '_async' if async_pass else ''
454                # awt = 'await ' if async_pass else ''
455                awt = ''
456                how = 'asynchronously' if async_pass else 'synchronously'
457
458                if len(msgtypes) == 1:
459                    # Special case: with a single message types we don't
460                    # use overloads.
461                    msgtype = msgtypes[0]
462                    msgtypevar = msgtype.__name__
463                    rtypes = msgtype.get_response_types()
464                    if len(rtypes) > 1:
465                        rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
466                    else:
467                        rtypevar = _filt_tp_name(rtypes[0])
468                    if async_pass:
469                        rtypevar = f'Awaitable[{rtypevar}]'
470                    out += (
471                        f'\n'
472                        f'    def send{sfx}(self,'
473                        f' message: {msgtypevar})'
474                        f' -> {rtypevar}:\n'
475                        f'        """Send a message {how}."""\n'
476                        f'        out = {awt}self._sender.'
477                        f'send{sfx}(self._obj, message)\n'
478                    )
479                    if not async_pass:
480                        out += (
481                            f'        assert isinstance(out, {rtypevar})\n'
482                            '        return out\n'
483                        )
484                    else:
485                        out += f'        return cast({rtypevar}, out)\n'
486
487                else:
488                    for msgtype in msgtypes:
489                        msgtypevar = msgtype.__name__
490                        rtypes = msgtype.get_response_types()
491                        if len(rtypes) > 1:
492                            rtypevar = ' | '.join(
493                                _filt_tp_name(t) for t in rtypes
494                            )
495                        else:
496                            rtypevar = _filt_tp_name(rtypes[0])
497                        out += (
498                            f'\n'
499                            f'    @overload\n'
500                            f'    {pfx}def send{sfx}(self,'
501                            f' message: {msgtypevar})'
502                            f' -> {rtypevar}:\n'
503                            f'        ...\n'
504                        )
505                    rtypevar = 'Response | None'
506                    if async_pass:
507                        rtypevar = f'Awaitable[{rtypevar}]'
508                    out += (
509                        f'\n'
510                        f'    def send{sfx}(self, message: Message)'
511                        f' -> {rtypevar}:\n'
512                        f'        """Send a message {how}."""\n'
513                        f'        return {awt}self._sender.'
514                        f'send{sfx}(self._obj, message)\n'
515                    )
516
517        return out
518
519    def do_create_receiver_module(
520        self,
521        basename: str,
522        protocol_create_code: str,
523        is_async: bool,
524        private: bool = False,
525        protocol_module_level_import_code: str | None = None,
526    ) -> str:
527        """Used by create_receiver_module(); do not call directly."""
528        # pylint: disable=too-many-locals
529        import textwrap
530
531        desc = 'asynchronous' if is_async else 'synchronous'
532        ppre = '_' if private else ''
533        msgtypes = list(self.message_ids_by_type.keys())
534        out = self._get_module_header(
535            'receiver',
536            extra_import_code=protocol_module_level_import_code,
537            enable_async_sends=False,
538        )
539        ccind = textwrap.indent(protocol_create_code, '        ')
540        out += (
541            f'class {ppre}{basename}(MessageReceiver):\n'
542            f'    """Protocol-specific {desc} receiver."""\n'
543            f'\n'
544            f'    is_async = {is_async}\n'
545            f'\n'
546            f'    def __init__(self) -> None:\n'
547            f'{ccind}\n'
548            f'        super().__init__(protocol)\n'
549            f'\n'
550            f'    def __get__(\n'
551            f'        self,\n'
552            f'        obj: Any,\n'
553            f'        type_in: Any = None,\n'
554            f'    ) -> {ppre}Bound{basename}:\n'
555            f'        return {ppre}Bound{basename}('
556            f'obj, self)\n'
557        )
558
559        # Define handler() overloads for all registered message types.
560
561        def _filt_tp_name(rtype: type[Response] | None) -> str:
562            return 'None' if rtype is None else rtype.__name__
563
564        if msgtypes:
565            cbgn = 'Awaitable[' if is_async else ''
566            cend = ']' if is_async else ''
567            if len(msgtypes) == 1:
568                # Special case: when we have a single message type we don't
569                # use overloads.
570                msgtype = msgtypes[0]
571                msgtypevar = msgtype.__name__
572                rtypes = msgtype.get_response_types()
573                if len(rtypes) > 1:
574                    rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
575                else:
576                    rtypevar = _filt_tp_name(rtypes[0])
577                rtypevar = f'{cbgn}{rtypevar}{cend}'
578                out += (
579                    f'\n'
580                    f'    def handler(\n'
581                    f'        self,\n'
582                    f'        call: Callable[[Any, {msgtypevar}], '
583                    f'{rtypevar}],\n'
584                    f'    )'
585                    f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n'
586                    f'        """Decorator to register message handlers."""\n'
587                    f'        from typing import cast, Callable, Any\n'
588                    f'\n'
589                    f'        self.register_handler(cast(Callable'
590                    f'[[Any, Message], Response], call))\n'
591                    f'        return call\n'
592                )
593            else:
594                for msgtype in msgtypes:
595                    msgtypevar = msgtype.__name__
596                    rtypes = msgtype.get_response_types()
597                    if len(rtypes) > 1:
598                        rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
599                    else:
600                        rtypevar = _filt_tp_name(rtypes[0])
601                    rtypevar = f'{cbgn}{rtypevar}{cend}'
602                    out += (
603                        f'\n'
604                        f'    @overload\n'
605                        f'    def handler(\n'
606                        f'        self,\n'
607                        f'        call: Callable[[Any, {msgtypevar}], '
608                        f'{rtypevar}],\n'
609                        f'    )'
610                        f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n'
611                        f'        ...\n'
612                    )
613                out += (
614                    '\n'
615                    '    def handler(self, call: Callable) -> Callable:\n'
616                    '        """Decorator to register message handlers."""\n'
617                    '        self.register_handler(call)\n'
618                    '        return call\n'
619                )
620
621        out += (
622            f'\n'
623            f'\n'
624            f'class {ppre}Bound{basename}(BoundMessageReceiver):\n'
625            f'    """Protocol-specific bound receiver."""\n'
626        )
627        if is_async:
628            out += (
629                '\n'
630                '    def handle_raw_message(\n'
631                '        self, message: str, raise_unregistered: bool = False\n'
632                '    ) -> Awaitable[str]:\n'
633                '        """Asynchronously handle a raw incoming message."""\n'
634                '        return self._receiver.'
635                'handle_raw_message_async(\n'
636                '            self._obj, message, raise_unregistered\n'
637                '        )\n'
638            )
639
640        else:
641            out += (
642                '\n'
643                '    def handle_raw_message(\n'
644                '        self, message: str, raise_unregistered: bool = False\n'
645                '    ) -> str:\n'
646                '        """Synchronously handle a raw incoming message."""\n'
647                '        return self._receiver.handle_raw_message(\n'
648                '            self._obj, message, raise_unregistered\n'
649                '        )\n'
650            )
651
652        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_remote_errors: 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_remote_errors: 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
 67        end - at least 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_remote_errors' 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_remote_errors = log_remote_errors

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_remote_errors' 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_remote_errors
@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                    traceback.format_exc()
218                    if self.remote_errors_include_stack_traces
219                    else 'An internal error has occurred.'
220                ),
221                error_type=ErrorSysResponse.ErrorType.REMOTE,
222            ),
223            self.log_remote_errors,
224        )

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:
240    @staticmethod
241    def decode_dict(data: str) -> dict:
242        """Decode data to a dict."""
243        out = json.loads(data)
244        assert isinstance(out, dict)
245        return out

Decode data to a dict.

def message_from_dict(self, data: dict) -> Message:
247    def message_from_dict(self, data: dict) -> Message:
248        """Decode a message from a json string."""
249        out = self._from_dict(data, self.message_types_by_id, 'message')
250        assert isinstance(out, Message)
251        return out

Decode a message from a json string.

def response_from_dict( self, data: dict) -> Response | SysResponse:
253    def response_from_dict(self, data: dict) -> Response | SysResponse:
254        """Decode a response from a json string."""
255        out = self._from_dict(data, self.response_types_by_id, 'response')
256        assert isinstance(out, Response | SysResponse)
257        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:
401    def do_create_sender_module(
402        self,
403        basename: str,
404        protocol_create_code: str,
405        enable_sync_sends: bool,
406        enable_async_sends: bool,
407        private: bool = False,
408        protocol_module_level_import_code: str | None = None,
409    ) -> str:
410        """Used by create_sender_module(); do not call directly."""
411        # pylint: disable=too-many-locals
412        # pylint: disable=too-many-branches
413        import textwrap
414
415        msgtypes = list(self.message_ids_by_type.keys())
416
417        ppre = '_' if private else ''
418        out = self._get_module_header(
419            'sender',
420            extra_import_code=protocol_module_level_import_code,
421            enable_async_sends=enable_async_sends,
422        )
423        ccind = textwrap.indent(protocol_create_code, '        ')
424        out += (
425            f'class {ppre}{basename}(MessageSender):\n'
426            f'    """Protocol-specific sender."""\n'
427            f'\n'
428            f'    def __init__(self) -> None:\n'
429            f'{ccind}\n'
430            f'        super().__init__(protocol)\n'
431            f'\n'
432            f'    def __get__(\n'
433            f'        self, obj: Any, type_in: Any = None\n'
434            f'    ) -> {ppre}Bound{basename}:\n'
435            f'        return {ppre}Bound{basename}(obj, self)\n'
436            f'\n'
437            f'\n'
438            f'class {ppre}Bound{basename}(BoundMessageSender):\n'
439            f'    """Protocol-specific bound sender."""\n'
440        )
441
442        def _filt_tp_name(rtype: type[Response] | None) -> str:
443            return 'None' if rtype is None else rtype.__name__
444
445        # Define handler() overloads for all registered message types.
446        if msgtypes:
447            for async_pass in False, True:
448                if async_pass and not enable_async_sends:
449                    continue
450                if not async_pass and not enable_sync_sends:
451                    continue
452                pfx = 'async ' if async_pass else ''
453                sfx = '_async' if async_pass else ''
454                # awt = 'await ' if async_pass else ''
455                awt = ''
456                how = 'asynchronously' if async_pass else 'synchronously'
457
458                if len(msgtypes) == 1:
459                    # Special case: with a single message types we don't
460                    # use overloads.
461                    msgtype = msgtypes[0]
462                    msgtypevar = msgtype.__name__
463                    rtypes = msgtype.get_response_types()
464                    if len(rtypes) > 1:
465                        rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
466                    else:
467                        rtypevar = _filt_tp_name(rtypes[0])
468                    if async_pass:
469                        rtypevar = f'Awaitable[{rtypevar}]'
470                    out += (
471                        f'\n'
472                        f'    def send{sfx}(self,'
473                        f' message: {msgtypevar})'
474                        f' -> {rtypevar}:\n'
475                        f'        """Send a message {how}."""\n'
476                        f'        out = {awt}self._sender.'
477                        f'send{sfx}(self._obj, message)\n'
478                    )
479                    if not async_pass:
480                        out += (
481                            f'        assert isinstance(out, {rtypevar})\n'
482                            '        return out\n'
483                        )
484                    else:
485                        out += f'        return cast({rtypevar}, out)\n'
486
487                else:
488                    for msgtype in msgtypes:
489                        msgtypevar = msgtype.__name__
490                        rtypes = msgtype.get_response_types()
491                        if len(rtypes) > 1:
492                            rtypevar = ' | '.join(
493                                _filt_tp_name(t) for t in rtypes
494                            )
495                        else:
496                            rtypevar = _filt_tp_name(rtypes[0])
497                        out += (
498                            f'\n'
499                            f'    @overload\n'
500                            f'    {pfx}def send{sfx}(self,'
501                            f' message: {msgtypevar})'
502                            f' -> {rtypevar}:\n'
503                            f'        ...\n'
504                        )
505                    rtypevar = 'Response | None'
506                    if async_pass:
507                        rtypevar = f'Awaitable[{rtypevar}]'
508                    out += (
509                        f'\n'
510                        f'    def send{sfx}(self, message: Message)'
511                        f' -> {rtypevar}:\n'
512                        f'        """Send a message {how}."""\n'
513                        f'        return {awt}self._sender.'
514                        f'send{sfx}(self._obj, message)\n'
515                    )
516
517        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:
519    def do_create_receiver_module(
520        self,
521        basename: str,
522        protocol_create_code: str,
523        is_async: bool,
524        private: bool = False,
525        protocol_module_level_import_code: str | None = None,
526    ) -> str:
527        """Used by create_receiver_module(); do not call directly."""
528        # pylint: disable=too-many-locals
529        import textwrap
530
531        desc = 'asynchronous' if is_async else 'synchronous'
532        ppre = '_' if private else ''
533        msgtypes = list(self.message_ids_by_type.keys())
534        out = self._get_module_header(
535            'receiver',
536            extra_import_code=protocol_module_level_import_code,
537            enable_async_sends=False,
538        )
539        ccind = textwrap.indent(protocol_create_code, '        ')
540        out += (
541            f'class {ppre}{basename}(MessageReceiver):\n'
542            f'    """Protocol-specific {desc} receiver."""\n'
543            f'\n'
544            f'    is_async = {is_async}\n'
545            f'\n'
546            f'    def __init__(self) -> None:\n'
547            f'{ccind}\n'
548            f'        super().__init__(protocol)\n'
549            f'\n'
550            f'    def __get__(\n'
551            f'        self,\n'
552            f'        obj: Any,\n'
553            f'        type_in: Any = None,\n'
554            f'    ) -> {ppre}Bound{basename}:\n'
555            f'        return {ppre}Bound{basename}('
556            f'obj, self)\n'
557        )
558
559        # Define handler() overloads for all registered message types.
560
561        def _filt_tp_name(rtype: type[Response] | None) -> str:
562            return 'None' if rtype is None else rtype.__name__
563
564        if msgtypes:
565            cbgn = 'Awaitable[' if is_async else ''
566            cend = ']' if is_async else ''
567            if len(msgtypes) == 1:
568                # Special case: when we have a single message type we don't
569                # use overloads.
570                msgtype = msgtypes[0]
571                msgtypevar = msgtype.__name__
572                rtypes = msgtype.get_response_types()
573                if len(rtypes) > 1:
574                    rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
575                else:
576                    rtypevar = _filt_tp_name(rtypes[0])
577                rtypevar = f'{cbgn}{rtypevar}{cend}'
578                out += (
579                    f'\n'
580                    f'    def handler(\n'
581                    f'        self,\n'
582                    f'        call: Callable[[Any, {msgtypevar}], '
583                    f'{rtypevar}],\n'
584                    f'    )'
585                    f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n'
586                    f'        """Decorator to register message handlers."""\n'
587                    f'        from typing import cast, Callable, Any\n'
588                    f'\n'
589                    f'        self.register_handler(cast(Callable'
590                    f'[[Any, Message], Response], call))\n'
591                    f'        return call\n'
592                )
593            else:
594                for msgtype in msgtypes:
595                    msgtypevar = msgtype.__name__
596                    rtypes = msgtype.get_response_types()
597                    if len(rtypes) > 1:
598                        rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
599                    else:
600                        rtypevar = _filt_tp_name(rtypes[0])
601                    rtypevar = f'{cbgn}{rtypevar}{cend}'
602                    out += (
603                        f'\n'
604                        f'    @overload\n'
605                        f'    def handler(\n'
606                        f'        self,\n'
607                        f'        call: Callable[[Any, {msgtypevar}], '
608                        f'{rtypevar}],\n'
609                        f'    )'
610                        f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n'
611                        f'        ...\n'
612                    )
613                out += (
614                    '\n'
615                    '    def handler(self, call: Callable) -> Callable:\n'
616                    '        """Decorator to register message handlers."""\n'
617                    '        self.register_handler(call)\n'
618                    '        return call\n'
619                )
620
621        out += (
622            f'\n'
623            f'\n'
624            f'class {ppre}Bound{basename}(BoundMessageReceiver):\n'
625            f'    """Protocol-specific bound receiver."""\n'
626        )
627        if is_async:
628            out += (
629                '\n'
630                '    def handle_raw_message(\n'
631                '        self, message: str, raise_unregistered: bool = False\n'
632                '    ) -> Awaitable[str]:\n'
633                '        """Asynchronously handle a raw incoming message."""\n'
634                '        return self._receiver.'
635                'handle_raw_message_async(\n'
636                '            self._obj, message, raise_unregistered\n'
637                '        )\n'
638            )
639
640        else:
641            out += (
642                '\n'
643                '    def handle_raw_message(\n'
644                '        self, message: str, raise_unregistered: bool = False\n'
645                '    ) -> str:\n'
646                '        """Synchronously handle a raw incoming message."""\n'
647                '        return self._receiver.handle_raw_message(\n'
648                '            self._obj, message, raise_unregistered\n'
649                '        )\n'
650            )
651
652        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    This is instantiated at the class level and used to register unbound
 25    class methods to handle raw message sending.
 26
 27    Example:
 28
 29    class MyClass:
 30        msg = MyMessageSender(some_protocol)
 31
 32        @msg.send_method
 33        def send_raw_message(self, message: str) -> str:
 34            # Actually send the message here.
 35
 36    # MyMessageSender class should provide overloads for send(), send_async(),
 37    # etc. to ensure all sending happens with valid types.
 38    obj = MyClass()
 39    obj.msg.send(SomeMessageType())
 40    """
 41
 42    def __init__(self, protocol: MessageProtocol) -> None:
 43        self.protocol = protocol
 44        self._send_raw_message_call: Callable[[Any, str], str] | None = None
 45        self._send_async_raw_message_call: Callable[
 46            [Any, str], Awaitable[str]
 47        ] | None = None
 48        self._send_async_raw_message_ex_call: Callable[
 49            [Any, str, Message], Awaitable[str]
 50        ] | None = None
 51        self._encode_filter_call: Callable[
 52            [Any, Message, dict], None
 53        ] | None = None
 54        self._decode_filter_call: Callable[
 55            [Any, Message, dict, Response | SysResponse], None
 56        ] | None = None
 57        self._peer_desc_call: Callable[[Any], str] | None = None
 58
 59    def send_method(
 60        self, call: Callable[[Any, str], str]
 61    ) -> Callable[[Any, str], str]:
 62        """Function decorator for setting raw send method.
 63
 64        Send methods take strings and should return strings.
 65        CommunicationErrors raised here will be returned to the sender
 66        as such; all other exceptions will result in a RuntimeError for
 67        the sender.
 68        """
 69        assert self._send_raw_message_call is None
 70        self._send_raw_message_call = call
 71        return call
 72
 73    def send_async_method(
 74        self, call: Callable[[Any, str], Awaitable[str]]
 75    ) -> Callable[[Any, str], Awaitable[str]]:
 76        """Function decorator for setting raw send-async method.
 77
 78        Send methods take strings and should return strings.
 79        CommunicationErrors raised here will be returned to the sender
 80        as such; all other exceptions will result in a RuntimeError for
 81        the sender.
 82
 83        IMPORTANT: Generally async send methods should not be implemented
 84        as 'async' methods, but instead should be regular methods that
 85        return awaitable objects. This way it can be guaranteed that
 86        outgoing messages are synchronously enqueued in the correct
 87        order, and then async calls can be returned which finish each
 88        send. If the entire call is async, they may be enqueued out of
 89        order in rare cases.
 90        """
 91        assert self._send_async_raw_message_call is None
 92        self._send_async_raw_message_call = call
 93        return call
 94
 95    def send_async_ex_method(
 96        self, call: Callable[[Any, str, Message], Awaitable[str]]
 97    ) -> Callable[[Any, str, Message], Awaitable[str]]:
 98        """Function decorator for extended send-async method.
 99
100        Version of send_async_method which is also is passed the original
101        unencoded message; can be useful for cases where metadata is sent
102        along with messages referring to their payloads/etc.
103        """
104        assert self._send_async_raw_message_ex_call is None
105        self._send_async_raw_message_ex_call = call
106        return call
107
108    def encode_filter_method(
109        self, call: Callable[[Any, Message, dict], None]
110    ) -> Callable[[Any, Message, dict], None]:
111        """Function decorator for defining an encode filter.
112
113        Encode filters can be used to add extra data to the message
114        dict before is is encoded to a string and sent out.
115        """
116        assert self._encode_filter_call is None
117        self._encode_filter_call = call
118        return call
119
120    def decode_filter_method(
121        self, call: Callable[[Any, Message, dict, Response | SysResponse], None]
122    ) -> Callable[[Any, Message, dict, Response], None]:
123        """Function decorator for defining a decode filter.
124
125        Decode filters can be used to extract extra data from incoming
126        message dicts.
127        """
128        assert self._decode_filter_call is None
129        self._decode_filter_call = call
130        return call
131
132    def peer_desc_method(
133        self, call: Callable[[Any], str]
134    ) -> Callable[[Any], str]:
135        """Function decorator for defining peer descriptions.
136
137        These are included in error messages or other diagnostics.
138        """
139        assert self._peer_desc_call is None
140        self._peer_desc_call = call
141        return call
142
143    def send(self, bound_obj: Any, message: Message) -> Response | None:
144        """Send a message synchronously."""
145        return self.unpack_raw_response(
146            bound_obj=bound_obj,
147            message=message,
148            raw_response=self.fetch_raw_response(
149                bound_obj=bound_obj,
150                message=message,
151            ),
152        )
153
154    def send_async(
155        self, bound_obj: Any, message: Message
156    ) -> Awaitable[Response | None]:
157        """Send a message asynchronously."""
158
159        # Note: This call is synchronous so that the first part of it can
160        # happen synchronously. If the whole call were async we wouldn't be
161        # able to guarantee that messages sent in order would actually go
162        # out in order.
163        raw_response_awaitable = self.fetch_raw_response_async(
164            bound_obj=bound_obj,
165            message=message,
166        )
167        # Now return an awaitable that will finish the send.
168        return self._send_async_awaitable(
169            bound_obj, message, raw_response_awaitable
170        )
171
172    async def _send_async_awaitable(
173        self,
174        bound_obj: Any,
175        message: Message,
176        raw_response_awaitable: Awaitable[Response | SysResponse],
177    ) -> Response | None:
178        return self.unpack_raw_response(
179            bound_obj=bound_obj,
180            message=message,
181            raw_response=await raw_response_awaitable,
182        )
183
184    def fetch_raw_response(
185        self, bound_obj: Any, message: Message
186    ) -> Response | SysResponse:
187        """Send a message synchronously.
188
189        Generally you can just call send(); these split versions are
190        for when message sending and response handling need to happen
191        in different contexts/threads.
192        """
193        if self._send_raw_message_call is None:
194            raise RuntimeError('send() is unimplemented for this type.')
195
196        msg_encoded = self._encode_message(bound_obj, message)
197        try:
198            response_encoded = self._send_raw_message_call(
199                bound_obj, msg_encoded
200            )
201        except Exception as exc:
202            response = ErrorSysResponse(
203                error_message='Error in MessageSender @send_method.',
204                error_type=(
205                    ErrorSysResponse.ErrorType.COMMUNICATION
206                    if isinstance(exc, CommunicationError)
207                    else ErrorSysResponse.ErrorType.LOCAL
208                ),
209            )
210            # Can include the actual exception since we'll be looking at
211            # this locally; might be helpful.
212            response.set_local_exception(exc)
213            return response
214        return self._decode_raw_response(bound_obj, message, response_encoded)
215
216    def fetch_raw_response_async(
217        self, bound_obj: Any, message: Message
218    ) -> Awaitable[Response | SysResponse]:
219        """Fetch a raw message response awaitable.
220
221        The result of this should be awaited and then passed to
222        unpack_raw_response() to produce the final message result.
223
224        Generally you can just call send(); calling fetch and unpack
225        manually is for when message sending and response handling need
226        to happen in different contexts/threads.
227        """
228
229        # Note: This call is synchronous so that the first part of it can
230        # happen synchronously. If the whole call were async we wouldn't be
231        # able to guarantee that messages sent in order would actually go
232        # out in order.
233        if (
234            self._send_async_raw_message_call is None
235            and self._send_async_raw_message_ex_call is None
236        ):
237            raise RuntimeError('send_async() is unimplemented for this type.')
238
239        msg_encoded = self._encode_message(bound_obj, message)
240        try:
241            if self._send_async_raw_message_ex_call is not None:
242                send_awaitable = self._send_async_raw_message_ex_call(
243                    bound_obj, msg_encoded, message
244                )
245            else:
246                assert self._send_async_raw_message_call is not None
247                send_awaitable = self._send_async_raw_message_call(
248                    bound_obj, msg_encoded
249                )
250        except Exception as exc:
251            return self._error_awaitable(exc)
252
253        # Now return an awaitable to finish the job.
254        return self._fetch_raw_response_awaitable(
255            bound_obj, message, send_awaitable
256        )
257
258    async def _error_awaitable(self, exc: Exception) -> SysResponse:
259        response = ErrorSysResponse(
260            error_message='Error in MessageSender @send_async_method.',
261            error_type=(
262                ErrorSysResponse.ErrorType.COMMUNICATION
263                if isinstance(exc, CommunicationError)
264                else ErrorSysResponse.ErrorType.LOCAL
265            ),
266        )
267        # Can include the actual exception since we'll be looking at
268        # this locally; might be helpful.
269        response.set_local_exception(exc)
270        return response
271
272    async def _fetch_raw_response_awaitable(
273        self, bound_obj: Any, message: Message, send_awaitable: Awaitable[str]
274    ) -> Response | SysResponse:
275        try:
276            response_encoded = await send_awaitable
277        except Exception as exc:
278            response = ErrorSysResponse(
279                error_message='Error in MessageSender @send_async_method.',
280                error_type=(
281                    ErrorSysResponse.ErrorType.COMMUNICATION
282                    if isinstance(exc, CommunicationError)
283                    else ErrorSysResponse.ErrorType.LOCAL
284                ),
285            )
286            # Can include the actual exception since we'll be looking at
287            # this locally; might be helpful.
288            response.set_local_exception(exc)
289            return response
290        return self._decode_raw_response(bound_obj, message, response_encoded)
291
292    def unpack_raw_response(
293        self,
294        bound_obj: Any,
295        message: Message,
296        raw_response: Response | SysResponse,
297    ) -> Response | None:
298        """Convert a raw fetched response into a final response/error/etc.
299
300        Generally you can just call send(); calling fetch and unpack
301        manually is for when message sending and response handling need
302        to happen in different contexts/threads.
303        """
304        response = self._unpack_raw_response(bound_obj, raw_response)
305        assert (
306            response is None
307            or type(response) in type(message).get_response_types()
308        )
309        return response
310
311    def _encode_message(self, bound_obj: Any, message: Message) -> str:
312        """Encode a message for sending."""
313        msg_dict = self.protocol.message_to_dict(message)
314        if self._encode_filter_call is not None:
315            self._encode_filter_call(bound_obj, message, msg_dict)
316        return self.protocol.encode_dict(msg_dict)
317
318    def _decode_raw_response(
319        self, bound_obj: Any, message: Message, response_encoded: str
320    ) -> Response | SysResponse:
321        """Create a Response from returned data.
322
323        These Responses may encapsulate things like remote errors and
324        should not be handed directly to users. _unpack_raw_response()
325        should be used to translate to special values like None or raise
326        Exceptions. This function itself should never raise Exceptions.
327        """
328        response: Response | SysResponse
329        try:
330            response_dict = self.protocol.decode_dict(response_encoded)
331            response = self.protocol.response_from_dict(response_dict)
332            if self._decode_filter_call is not None:
333                self._decode_filter_call(
334                    bound_obj, message, response_dict, response
335                )
336        except Exception as exc:
337            response = ErrorSysResponse(
338                error_message='Error decoding raw response.',
339                error_type=ErrorSysResponse.ErrorType.LOCAL,
340            )
341            # Since we'll be looking at this locally, we can include
342            # extra info for logging/etc.
343            response.set_local_exception(exc)
344        return response
345
346    def _unpack_raw_response(
347        self, bound_obj: Any, raw_response: Response | SysResponse
348    ) -> Response | None:
349        """Given a raw Response, unpacks to special values or Exceptions.
350
351        The result of this call is what should be passed to users.
352        For complex messaging situations such as response callbacks
353        operating across different threads, this last stage should be
354        run such that any raised Exception is active when the callback
355        fires; not on the thread where the message was sent.
356        """
357        # EmptySysResponse translates to None
358        if isinstance(raw_response, EmptySysResponse):
359            return None
360
361        # Some error occurred. Raise a local Exception for it.
362        if isinstance(raw_response, ErrorSysResponse):
363            # Errors that happened locally can attach their exceptions
364            # here for extra logging goodness.
365            local_exception = raw_response.get_local_exception()
366
367            if (
368                raw_response.error_type
369                is ErrorSysResponse.ErrorType.COMMUNICATION
370            ):
371                raise CommunicationError(
372                    raw_response.error_message
373                ) from local_exception
374
375            # If something went wrong on *our* end of the connection,
376            # don't say it was a remote error.
377            if raw_response.error_type is ErrorSysResponse.ErrorType.LOCAL:
378                raise RuntimeError(
379                    raw_response.error_message
380                ) from local_exception
381
382            # If they want to support clean errors, do those.
383            if (
384                self.protocol.forward_clean_errors
385                and raw_response.error_type
386                is ErrorSysResponse.ErrorType.REMOTE_CLEAN
387            ):
388                raise CleanError(
389                    raw_response.error_message
390                ) from local_exception
391
392            if (
393                self.protocol.forward_communication_errors
394                and raw_response.error_type
395                is ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION
396            ):
397                raise CommunicationError(
398                    raw_response.error_message
399                ) from local_exception
400
401            # Everything else gets lumped in as a remote error.
402            raise RemoteError(
403                raw_response.error_message,
404                peer_desc=(
405                    'peer'
406                    if self._peer_desc_call is None
407                    else self._peer_desc_call(bound_obj)
408                ),
409            ) from local_exception
410
411        assert isinstance(raw_response, Response)
412        return raw_response

Facilitates sending messages to a target and receiving responses. This is instantiated at the class level and used to register unbound class methods to handle raw message sending.

Example:

class MyClass: msg = MyMessageSender(some_protocol)

@msg.send_method
def send_raw_message(self, message: str) -> str:
    # Actually send the message here.

MyMessageSender class should provide overloads for send(), send_async(),

etc. to ensure all sending happens with valid types.

obj = MyClass() obj.msg.send(SomeMessageType())

MessageSender(protocol: MessageProtocol)
42    def __init__(self, protocol: MessageProtocol) -> None:
43        self.protocol = protocol
44        self._send_raw_message_call: Callable[[Any, str], str] | None = None
45        self._send_async_raw_message_call: Callable[
46            [Any, str], Awaitable[str]
47        ] | None = None
48        self._send_async_raw_message_ex_call: Callable[
49            [Any, str, Message], Awaitable[str]
50        ] | None = None
51        self._encode_filter_call: Callable[
52            [Any, Message, dict], None
53        ] | None = None
54        self._decode_filter_call: Callable[
55            [Any, Message, dict, Response | SysResponse], None
56        ] | None = None
57        self._peer_desc_call: Callable[[Any], str] | None = None
protocol
def send_method(self, call: Callable[[Any, str], str]) -> Callable[[Any, str], str]:
59    def send_method(
60        self, call: Callable[[Any, str], str]
61    ) -> Callable[[Any, str], str]:
62        """Function decorator for setting raw send method.
63
64        Send methods take strings and should return strings.
65        CommunicationErrors raised here will be returned to the sender
66        as such; all other exceptions will result in a RuntimeError for
67        the sender.
68        """
69        assert self._send_raw_message_call is None
70        self._send_raw_message_call = call
71        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]]:
73    def send_async_method(
74        self, call: Callable[[Any, str], Awaitable[str]]
75    ) -> Callable[[Any, str], Awaitable[str]]:
76        """Function decorator for setting raw send-async method.
77
78        Send methods take strings and should return strings.
79        CommunicationErrors raised here will be returned to the sender
80        as such; all other exceptions will result in a RuntimeError for
81        the sender.
82
83        IMPORTANT: Generally async send methods should not be implemented
84        as 'async' methods, but instead should be regular methods that
85        return awaitable objects. This way it can be guaranteed that
86        outgoing messages are synchronously enqueued in the correct
87        order, and then async calls can be returned which finish each
88        send. If the entire call is async, they may be enqueued out of
89        order in rare cases.
90        """
91        assert self._send_async_raw_message_call is None
92        self._send_async_raw_message_call = call
93        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]]:
 95    def send_async_ex_method(
 96        self, call: Callable[[Any, str, Message], Awaitable[str]]
 97    ) -> Callable[[Any, str, Message], Awaitable[str]]:
 98        """Function decorator for extended send-async method.
 99
100        Version of send_async_method which is also is passed the original
101        unencoded message; can be useful for cases where metadata is sent
102        along with messages referring to their payloads/etc.
103        """
104        assert self._send_async_raw_message_ex_call is None
105        self._send_async_raw_message_ex_call = call
106        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]:
108    def encode_filter_method(
109        self, call: Callable[[Any, Message, dict], None]
110    ) -> Callable[[Any, Message, dict], None]:
111        """Function decorator for defining an encode filter.
112
113        Encode filters can be used to add extra data to the message
114        dict before is is encoded to a string and sent out.
115        """
116        assert self._encode_filter_call is None
117        self._encode_filter_call = call
118        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]:
120    def decode_filter_method(
121        self, call: Callable[[Any, Message, dict, Response | SysResponse], None]
122    ) -> Callable[[Any, Message, dict, Response], None]:
123        """Function decorator for defining a decode filter.
124
125        Decode filters can be used to extract extra data from incoming
126        message dicts.
127        """
128        assert self._decode_filter_call is None
129        self._decode_filter_call = call
130        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]:
132    def peer_desc_method(
133        self, call: Callable[[Any], str]
134    ) -> Callable[[Any], str]:
135        """Function decorator for defining peer descriptions.
136
137        These are included in error messages or other diagnostics.
138        """
139        assert self._peer_desc_call is None
140        self._peer_desc_call = call
141        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:
143    def send(self, bound_obj: Any, message: Message) -> Response | None:
144        """Send a message synchronously."""
145        return self.unpack_raw_response(
146            bound_obj=bound_obj,
147            message=message,
148            raw_response=self.fetch_raw_response(
149                bound_obj=bound_obj,
150                message=message,
151            ),
152        )

Send a message synchronously.

def send_async( self, bound_obj: Any, message: Message) -> Awaitable[Response | None]:
154    def send_async(
155        self, bound_obj: Any, message: Message
156    ) -> Awaitable[Response | None]:
157        """Send a message asynchronously."""
158
159        # Note: This call is synchronous so that the first part of it can
160        # happen synchronously. If the whole call were async we wouldn't be
161        # able to guarantee that messages sent in order would actually go
162        # out in order.
163        raw_response_awaitable = self.fetch_raw_response_async(
164            bound_obj=bound_obj,
165            message=message,
166        )
167        # Now return an awaitable that will finish the send.
168        return self._send_async_awaitable(
169            bound_obj, message, raw_response_awaitable
170        )

Send a message asynchronously.

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

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:
292    def unpack_raw_response(
293        self,
294        bound_obj: Any,
295        message: Message,
296        raw_response: Response | SysResponse,
297    ) -> Response | None:
298        """Convert a raw fetched response into a final response/error/etc.
299
300        Generally you can just call send(); calling fetch and unpack
301        manually is for when message sending and response handling need
302        to happen in different contexts/threads.
303        """
304        response = self._unpack_raw_response(bound_obj, raw_response)
305        assert (
306            response is None
307            or type(response) in type(message).get_response_types()
308        )
309        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:
415class BoundMessageSender:
416    """Base class for bound senders."""
417
418    def __init__(self, obj: Any, sender: MessageSender) -> None:
419        # Note: not checking obj here since we want to support
420        # at least our protocol property when accessed via type.
421        self._obj = obj
422        self._sender = sender
423
424    @property
425    def protocol(self) -> MessageProtocol:
426        """Protocol associated with this sender."""
427        return self._sender.protocol
428
429    def send_untyped(self, message: Message) -> Response | None:
430        """Send a message synchronously.
431
432        Whenever possible, use the send() call provided by generated
433        subclasses instead of this; it will provide better type safety.
434        """
435        assert self._obj is not None
436        return self._sender.send(bound_obj=self._obj, message=message)
437
438    def send_async_untyped(
439        self, message: Message
440    ) -> Awaitable[Response | None]:
441        """Send a message asynchronously.
442
443        Whenever possible, use the send_async() 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_async(bound_obj=self._obj, message=message)
448
449    def fetch_raw_response_async_untyped(
450        self, message: Message
451    ) -> Awaitable[Response | SysResponse]:
452        """Split send (part 1 of 2)."""
453        assert self._obj is not None
454        return self._sender.fetch_raw_response_async(
455            bound_obj=self._obj, message=message
456        )
457
458    def unpack_raw_response_untyped(
459        self, message: Message, raw_response: Response | SysResponse
460    ) -> Response | None:
461        """Split send (part 2 of 2)."""
462        return self._sender.unpack_raw_response(
463            bound_obj=self._obj, message=message, raw_response=raw_response
464        )

Base class for bound senders.

BoundMessageSender(obj: Any, sender: MessageSender)
418    def __init__(self, obj: Any, sender: MessageSender) -> None:
419        # Note: not checking obj here since we want to support
420        # at least our protocol property when accessed via type.
421        self._obj = obj
422        self._sender = sender
protocol: MessageProtocol

Protocol associated with this sender.

def send_untyped( self, message: Message) -> Response | None:
429    def send_untyped(self, message: Message) -> Response | None:
430        """Send a message synchronously.
431
432        Whenever possible, use the send() call provided by generated
433        subclasses instead of this; it will provide better type safety.
434        """
435        assert self._obj is not None
436        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]:
438    def send_async_untyped(
439        self, message: Message
440    ) -> Awaitable[Response | None]:
441        """Send a message asynchronously.
442
443        Whenever possible, use the send_async() 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_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]:
449    def fetch_raw_response_async_untyped(
450        self, message: Message
451    ) -> Awaitable[Response | SysResponse]:
452        """Split send (part 1 of 2)."""
453        assert self._obj is not None
454        return self._sender.fetch_raw_response_async(
455            bound_obj=self._obj, message=message
456        )

Split send (part 1 of 2).

def unpack_raw_response_untyped( self, message: Message, raw_response: Response | SysResponse) -> Response | None:
458    def unpack_raw_response_untyped(
459        self, message: Message, raw_response: Response | SysResponse
460    ) -> Response | None:
461        """Split send (part 2 of 2)."""
462        return self._sender.unpack_raw_response(
463            bound_obj=self._obj, message=message, raw_response=raw_response
464        )

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        @receiver.handler
 43        def handle_some_message_type(self, message: SomeMsg) -> SomeResponse:
 44            # Deal with this message type here.
 45
 46    # This will trigger the registered handler being called.
 47    obj = MyClass()
 48    obj.receiver.handle_raw_message(some_raw_data)
 49
 50    Any unhandled Exception occurring during message handling will result in
 51    an Exception being raised on the sending end.
 52    """
 53
 54    is_async = False
 55
 56    def __init__(self, protocol: MessageProtocol) -> None:
 57        self.protocol = protocol
 58        self._handlers: dict[type[Message], Callable] = {}
 59        self._decode_filter_call: Callable[
 60            [Any, dict, Message], None
 61        ] | None = None
 62        self._encode_filter_call: Callable[
 63            [Any, Message | None, Response | SysResponse, dict], None
 64        ] | None = None
 65
 66    # noinspection PyProtectedMember
 67    def register_handler(
 68        self, call: Callable[[Any, Message], Response | None]
 69    ) -> None:
 70        """Register a handler call.
 71
 72        The message type handled by the call is determined by its
 73        type annotation.
 74        """
 75        # TODO: can use types.GenericAlias in 3.9.
 76        # (hmm though now that we're there,  it seems a drop-in
 77        # replace gives us errors. Should re-test in 3.11 as it seems
 78        # that typing_extensions handles it differently in that case)
 79        from typing import _GenericAlias  # type: ignore
 80        from typing import get_type_hints, get_args
 81
 82        sig = inspect.getfullargspec(call)
 83
 84        # The provided callable should be a method taking one 'msg' arg.
 85        expectedsig = ['self', 'msg']
 86        if sig.args != expectedsig:
 87            raise ValueError(
 88                f'Expected callable signature of {expectedsig};'
 89                f' got {sig.args}'
 90            )
 91
 92        # Make sure we are only given async methods if we are an async handler
 93        # and sync ones otherwise.
 94        # UPDATE - can't do this anymore since we now sometimes use
 95        # regular functions which return awaitables instead of having
 96        # the entire function be async.
 97        # is_async = inspect.iscoroutinefunction(call)
 98        # if self.is_async != is_async:
 99        #     msg = (
100        #         'Expected a sync method; found an async one.'
101        #         if is_async
102        #         else 'Expected an async method; found a sync one.'
103        #     )
104        #     raise ValueError(msg)
105
106        # Check annotation types to determine what message types we handle.
107        # Return-type annotation can be a Union, but we probably don't
108        # have it available at runtime. Explicitly pull it in.
109        # UPDATE: we've updated our pylint filter to where we should
110        # have all annotations available.
111        # anns = get_type_hints(call, localns={'Union': Union})
112        anns = get_type_hints(call)
113
114        msgtype = anns.get('msg')
115        if not isinstance(msgtype, type):
116            raise TypeError(
117                f'expected a type for "msg" annotation; got {type(msgtype)}.'
118            )
119        assert issubclass(msgtype, Message)
120
121        ret = anns.get('return')
122        responsetypes: tuple[type[Any] | None, ...]
123
124        # Return types can be a single type or a union of types.
125        if isinstance(ret, (_GenericAlias, types.UnionType)):
126            targs = get_args(ret)
127            if not all(isinstance(a, (type, type(None))) for a in targs):
128                raise TypeError(
129                    f'expected only types for "return" annotation;'
130                    f' got {targs}.'
131                )
132            responsetypes = targs
133        else:
134            if not isinstance(ret, (type, type(None))):
135                raise TypeError(
136                    f'expected one or more types for'
137                    f' "return" annotation; got a {type(ret)}.'
138                )
139            # This seems like maybe a mypy bug. Appeared after adding
140            # types.UnionType above.
141            responsetypes = (ret,)
142
143        # This will contain NoneType for empty return cases, but
144        # we expect it to be None.
145        # noinspection PyPep8
146        responsetypes = tuple(
147            None if r is type(None) else r for r in responsetypes
148        )
149
150        # Make sure our protocol has this message type registered and our
151        # return types exactly match. (Technically we could return a subset
152        # of the supported types; can allow this in the future if it makes
153        # sense).
154        registered_types = self.protocol.message_ids_by_type.keys()
155
156        if msgtype not in registered_types:
157            raise TypeError(
158                f'Message type {msgtype} is not registered'
159                f' in this Protocol.'
160            )
161
162        if msgtype in self._handlers:
163            raise TypeError(
164                f'Message type {msgtype} already has a registered' f' handler.'
165            )
166
167        # Make sure the responses exactly matches what the message expects.
168        if set(responsetypes) != set(msgtype.get_response_types()):
169            raise TypeError(
170                f'Provided response types {responsetypes} do not'
171                f' match the set expected by message type {msgtype}: '
172                f'({msgtype.get_response_types()})'
173            )
174
175        # Ok; we're good!
176        self._handlers[msgtype] = call
177
178    def decode_filter_method(
179        self, call: Callable[[Any, dict, Message], None]
180    ) -> Callable[[Any, dict, Message], None]:
181        """Function decorator for defining a decode filter.
182
183        Decode filters can be used to extract extra data from incoming
184        message dicts. This version will work for both handle_raw_message()
185        and handle_raw_message_async()
186        """
187        assert self._decode_filter_call is None
188        self._decode_filter_call = call
189        return call
190
191    def encode_filter_method(
192        self,
193        call: Callable[
194            [Any, Message | None, Response | SysResponse, dict], None
195        ],
196    ) -> Callable[[Any, Message | None, Response, dict], None]:
197        """Function decorator for defining an encode filter.
198
199        Encode filters can be used to add extra data to the message
200        dict before is is encoded to a string and sent out.
201        """
202        assert self._encode_filter_call is None
203        self._encode_filter_call = call
204        return call
205
206    def validate(self, log_only: bool = False) -> None:
207        """Check for handler completeness, valid types, etc."""
208        for msgtype in self.protocol.message_ids_by_type.keys():
209            if issubclass(msgtype, Response):
210                continue
211            if msgtype not in self._handlers:
212                msg = (
213                    f'Protocol message type {msgtype} is not handled'
214                    f' by receiver type {type(self)}.'
215                )
216                if log_only:
217                    logging.error(msg)
218                else:
219                    raise TypeError(msg)
220
221    def _decode_incoming_message_base(
222        self, bound_obj: Any, msg: str
223    ) -> tuple[Any, dict, Message]:
224        # Decode the incoming message.
225        msg_dict = self.protocol.decode_dict(msg)
226        msg_decoded = self.protocol.message_from_dict(msg_dict)
227        assert isinstance(msg_decoded, Message)
228        if self._decode_filter_call is not None:
229            self._decode_filter_call(bound_obj, msg_dict, msg_decoded)
230        return bound_obj, msg_dict, msg_decoded
231
232    def _decode_incoming_message(self, bound_obj: Any, msg: str) -> Message:
233        bound_obj, _msg_dict, msg_decoded = self._decode_incoming_message_base(
234            bound_obj=bound_obj, msg=msg
235        )
236        return msg_decoded
237
238    def encode_user_response(
239        self, bound_obj: Any, message: Message, response: Response | None
240    ) -> str:
241        """Encode a response provided by the user for sending."""
242
243        assert isinstance(response, Response | None)
244        # (user should never explicitly return error-responses)
245        assert (
246            response is None or type(response) in message.get_response_types()
247        )
248
249        # A return value of None equals EmptySysResponse.
250        out_response: Response | SysResponse
251        if response is None:
252            out_response = EmptySysResponse()
253        else:
254            out_response = response
255
256        response_dict = self.protocol.response_to_dict(out_response)
257        if self._encode_filter_call is not None:
258            self._encode_filter_call(
259                bound_obj, message, out_response, response_dict
260            )
261        return self.protocol.encode_dict(response_dict)
262
263    def encode_error_response(
264        self, bound_obj: Any, message: Message | None, exc: Exception
265    ) -> tuple[str, bool]:
266        """Given an error, return sysresponse str and whether to log."""
267        response, dolog = self.protocol.error_to_response(exc)
268        response_dict = self.protocol.response_to_dict(response)
269        if self._encode_filter_call is not None:
270            self._encode_filter_call(
271                bound_obj, message, response, response_dict
272            )
273        return self.protocol.encode_dict(response_dict), dolog
274
275    def handle_raw_message(
276        self, bound_obj: Any, msg: str, raise_unregistered: bool = False
277    ) -> str:
278        """Decode, handle, and return an response for a message.
279
280        if 'raise_unregistered' is True, will raise an
281        efro.message.UnregisteredMessageIDError for messages not handled by
282        the protocol. In all other cases local errors will translate to
283        error responses returned to the sender.
284        """
285        assert not self.is_async, "can't call sync handler on async receiver"
286        msg_decoded: Message | None = None
287        msgtype: type[Message] | None = None
288        try:
289            msg_decoded = self._decode_incoming_message(bound_obj, msg)
290            msgtype = type(msg_decoded)
291            handler = self._handlers.get(msgtype)
292            if handler is None:
293                raise RuntimeError(f'Got unhandled message type: {msgtype}.')
294            response = handler(bound_obj, msg_decoded)
295            assert isinstance(response, Response | None)
296            return self.encode_user_response(bound_obj, msg_decoded, response)
297
298        except Exception as exc:
299            if raise_unregistered and isinstance(
300                exc, UnregisteredMessageIDError
301            ):
302                raise
303            rstr, dolog = self.encode_error_response(
304                bound_obj, msg_decoded, exc
305            )
306            if dolog:
307                if msgtype is not None:
308                    logging.exception(
309                        'Error handling %s.%s message.',
310                        msgtype.__module__,
311                        msgtype.__qualname__,
312                    )
313                else:
314                    logging.exception('Error in efro.message handling.')
315            return rstr
316
317    def handle_raw_message_async(
318        self, bound_obj: Any, msg: str, raise_unregistered: bool = False
319    ) -> Awaitable[str]:
320        """Should be called when the receiver gets a message.
321
322        The return value is the raw response to the message.
323        """
324
325        # Note: This call is synchronous so that the first part of it can
326        # happen synchronously. If the whole call were async we wouldn't be
327        # able to guarantee that messages handlers would be called in the
328        # order the messages were received.
329
330        assert self.is_async, "can't call async handler on sync receiver"
331        msg_decoded: Message | None = None
332        msgtype: type[Message] | None = None
333        try:
334            msg_decoded = self._decode_incoming_message(bound_obj, msg)
335            msgtype = type(msg_decoded)
336            handler = self._handlers.get(msgtype)
337            if handler is None:
338                raise RuntimeError(f'Got unhandled message type: {msgtype}.')
339            handler_awaitable = handler(bound_obj, msg_decoded)
340
341        except Exception as exc:
342            if raise_unregistered and isinstance(
343                exc, UnregisteredMessageIDError
344            ):
345                raise
346            return self._handle_raw_message_async_error(
347                bound_obj, msg_decoded, msgtype, exc
348            )
349
350        # Return an awaitable to handle the rest asynchronously.
351        return self._handle_raw_message_async(
352            bound_obj, msg_decoded, msgtype, handler_awaitable
353        )
354
355    async def _handle_raw_message_async_error(
356        self,
357        bound_obj: Any,
358        msg_decoded: Message | None,
359        msgtype: type[Message] | None,
360        exc: Exception,
361    ) -> str:
362        rstr, dolog = self.encode_error_response(bound_obj, msg_decoded, exc)
363        if dolog:
364            if msgtype is not None:
365                logging.exception(
366                    'Error handling %s.%s message.',
367                    msgtype.__module__,
368                    msgtype.__qualname__,
369                )
370            else:
371                logging.exception('Error in efro.message handling.')
372        return rstr
373
374    async def _handle_raw_message_async(
375        self,
376        bound_obj: Any,
377        msg_decoded: Message,
378        msgtype: type[Message] | None,
379        handler_awaitable: Awaitable[Response | None],
380    ) -> str:
381        """Should be called when the receiver gets a message.
382
383        The return value is the raw response to the message.
384        """
385        try:
386            response = await handler_awaitable
387            assert isinstance(response, Response | None)
388            return self.encode_user_response(bound_obj, msg_decoded, response)
389
390        except Exception as exc:
391            return await self._handle_raw_message_async_error(
392                bound_obj, msg_decoded, msgtype, exc
393            )

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 Exception being raised on the sending end.

MessageReceiver(protocol: MessageProtocol)
56    def __init__(self, protocol: MessageProtocol) -> None:
57        self.protocol = protocol
58        self._handlers: dict[type[Message], Callable] = {}
59        self._decode_filter_call: Callable[
60            [Any, dict, Message], None
61        ] | None = None
62        self._encode_filter_call: Callable[
63            [Any, Message | None, Response | SysResponse, dict], None
64        ] | None = None
is_async = False
protocol
def register_handler( self, call: Callable[[Any, Message], Response | None]) -> None:
 67    def register_handler(
 68        self, call: Callable[[Any, Message], Response | None]
 69    ) -> None:
 70        """Register a handler call.
 71
 72        The message type handled by the call is determined by its
 73        type annotation.
 74        """
 75        # TODO: can use types.GenericAlias in 3.9.
 76        # (hmm though now that we're there,  it seems a drop-in
 77        # replace gives us errors. Should re-test in 3.11 as it seems
 78        # that typing_extensions handles it differently in that case)
 79        from typing import _GenericAlias  # type: ignore
 80        from typing import get_type_hints, get_args
 81
 82        sig = inspect.getfullargspec(call)
 83
 84        # The provided callable should be a method taking one 'msg' arg.
 85        expectedsig = ['self', 'msg']
 86        if sig.args != expectedsig:
 87            raise ValueError(
 88                f'Expected callable signature of {expectedsig};'
 89                f' got {sig.args}'
 90            )
 91
 92        # Make sure we are only given async methods if we are an async handler
 93        # and sync ones otherwise.
 94        # UPDATE - can't do this anymore since we now sometimes use
 95        # regular functions which return awaitables instead of having
 96        # the entire function be async.
 97        # is_async = inspect.iscoroutinefunction(call)
 98        # if self.is_async != is_async:
 99        #     msg = (
100        #         'Expected a sync method; found an async one.'
101        #         if is_async
102        #         else 'Expected an async method; found a sync one.'
103        #     )
104        #     raise ValueError(msg)
105
106        # Check annotation types to determine what message types we handle.
107        # Return-type annotation can be a Union, but we probably don't
108        # have it available at runtime. Explicitly pull it in.
109        # UPDATE: we've updated our pylint filter to where we should
110        # have all annotations available.
111        # anns = get_type_hints(call, localns={'Union': Union})
112        anns = get_type_hints(call)
113
114        msgtype = anns.get('msg')
115        if not isinstance(msgtype, type):
116            raise TypeError(
117                f'expected a type for "msg" annotation; got {type(msgtype)}.'
118            )
119        assert issubclass(msgtype, Message)
120
121        ret = anns.get('return')
122        responsetypes: tuple[type[Any] | None, ...]
123
124        # Return types can be a single type or a union of types.
125        if isinstance(ret, (_GenericAlias, types.UnionType)):
126            targs = get_args(ret)
127            if not all(isinstance(a, (type, type(None))) for a in targs):
128                raise TypeError(
129                    f'expected only types for "return" annotation;'
130                    f' got {targs}.'
131                )
132            responsetypes = targs
133        else:
134            if not isinstance(ret, (type, type(None))):
135                raise TypeError(
136                    f'expected one or more types for'
137                    f' "return" annotation; got a {type(ret)}.'
138                )
139            # This seems like maybe a mypy bug. Appeared after adding
140            # types.UnionType above.
141            responsetypes = (ret,)
142
143        # This will contain NoneType for empty return cases, but
144        # we expect it to be None.
145        # noinspection PyPep8
146        responsetypes = tuple(
147            None if r is type(None) else r for r in responsetypes
148        )
149
150        # Make sure our protocol has this message type registered and our
151        # return types exactly match. (Technically we could return a subset
152        # of the supported types; can allow this in the future if it makes
153        # sense).
154        registered_types = self.protocol.message_ids_by_type.keys()
155
156        if msgtype not in registered_types:
157            raise TypeError(
158                f'Message type {msgtype} is not registered'
159                f' in this Protocol.'
160            )
161
162        if msgtype in self._handlers:
163            raise TypeError(
164                f'Message type {msgtype} already has a registered' f' handler.'
165            )
166
167        # Make sure the responses exactly matches what the message expects.
168        if set(responsetypes) != set(msgtype.get_response_types()):
169            raise TypeError(
170                f'Provided response types {responsetypes} do not'
171                f' match the set expected by message type {msgtype}: '
172                f'({msgtype.get_response_types()})'
173            )
174
175        # Ok; we're good!
176        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]:
178    def decode_filter_method(
179        self, call: Callable[[Any, dict, Message], None]
180    ) -> Callable[[Any, dict, Message], None]:
181        """Function decorator for defining a decode filter.
182
183        Decode filters can be used to extract extra data from incoming
184        message dicts. This version will work for both handle_raw_message()
185        and handle_raw_message_async()
186        """
187        assert self._decode_filter_call is None
188        self._decode_filter_call = call
189        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]:
191    def encode_filter_method(
192        self,
193        call: Callable[
194            [Any, Message | None, Response | SysResponse, dict], None
195        ],
196    ) -> Callable[[Any, Message | None, Response, dict], None]:
197        """Function decorator for defining an encode filter.
198
199        Encode filters can be used to add extra data to the message
200        dict before is is encoded to a string and sent out.
201        """
202        assert self._encode_filter_call is None
203        self._encode_filter_call = call
204        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:
206    def validate(self, log_only: bool = False) -> None:
207        """Check for handler completeness, valid types, etc."""
208        for msgtype in self.protocol.message_ids_by_type.keys():
209            if issubclass(msgtype, Response):
210                continue
211            if msgtype not in self._handlers:
212                msg = (
213                    f'Protocol message type {msgtype} is not handled'
214                    f' by receiver type {type(self)}.'
215                )
216                if log_only:
217                    logging.error(msg)
218                else:
219                    raise TypeError(msg)

Check for handler completeness, valid types, etc.

def encode_user_response( self, bound_obj: Any, message: Message, response: Response | None) -> str:
238    def encode_user_response(
239        self, bound_obj: Any, message: Message, response: Response | None
240    ) -> str:
241        """Encode a response provided by the user for sending."""
242
243        assert isinstance(response, Response | None)
244        # (user should never explicitly return error-responses)
245        assert (
246            response is None or type(response) in message.get_response_types()
247        )
248
249        # A return value of None equals EmptySysResponse.
250        out_response: Response | SysResponse
251        if response is None:
252            out_response = EmptySysResponse()
253        else:
254            out_response = response
255
256        response_dict = self.protocol.response_to_dict(out_response)
257        if self._encode_filter_call is not None:
258            self._encode_filter_call(
259                bound_obj, message, out_response, response_dict
260            )
261        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]:
263    def encode_error_response(
264        self, bound_obj: Any, message: Message | None, exc: Exception
265    ) -> tuple[str, bool]:
266        """Given an error, return sysresponse str and whether to log."""
267        response, dolog = self.protocol.error_to_response(exc)
268        response_dict = self.protocol.response_to_dict(response)
269        if self._encode_filter_call is not None:
270            self._encode_filter_call(
271                bound_obj, message, response, response_dict
272            )
273        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:
275    def handle_raw_message(
276        self, bound_obj: Any, msg: str, raise_unregistered: bool = False
277    ) -> str:
278        """Decode, handle, and return an response for a message.
279
280        if 'raise_unregistered' is True, will raise an
281        efro.message.UnregisteredMessageIDError for messages not handled by
282        the protocol. In all other cases local errors will translate to
283        error responses returned to the sender.
284        """
285        assert not self.is_async, "can't call sync handler on async receiver"
286        msg_decoded: Message | None = None
287        msgtype: type[Message] | None = None
288        try:
289            msg_decoded = self._decode_incoming_message(bound_obj, msg)
290            msgtype = type(msg_decoded)
291            handler = self._handlers.get(msgtype)
292            if handler is None:
293                raise RuntimeError(f'Got unhandled message type: {msgtype}.')
294            response = handler(bound_obj, msg_decoded)
295            assert isinstance(response, Response | None)
296            return self.encode_user_response(bound_obj, msg_decoded, response)
297
298        except Exception as exc:
299            if raise_unregistered and isinstance(
300                exc, UnregisteredMessageIDError
301            ):
302                raise
303            rstr, dolog = self.encode_error_response(
304                bound_obj, msg_decoded, exc
305            )
306            if dolog:
307                if msgtype is not None:
308                    logging.exception(
309                        'Error handling %s.%s message.',
310                        msgtype.__module__,
311                        msgtype.__qualname__,
312                    )
313                else:
314                    logging.exception('Error in efro.message handling.')
315            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]:
317    def handle_raw_message_async(
318        self, bound_obj: Any, msg: str, raise_unregistered: bool = False
319    ) -> Awaitable[str]:
320        """Should be called when the receiver gets a message.
321
322        The return value is the raw response to the message.
323        """
324
325        # Note: This call is synchronous so that the first part of it can
326        # happen synchronously. If the whole call were async we wouldn't be
327        # able to guarantee that messages handlers would be called in the
328        # order the messages were received.
329
330        assert self.is_async, "can't call async handler on sync receiver"
331        msg_decoded: Message | None = None
332        msgtype: type[Message] | None = None
333        try:
334            msg_decoded = self._decode_incoming_message(bound_obj, msg)
335            msgtype = type(msg_decoded)
336            handler = self._handlers.get(msgtype)
337            if handler is None:
338                raise RuntimeError(f'Got unhandled message type: {msgtype}.')
339            handler_awaitable = handler(bound_obj, msg_decoded)
340
341        except Exception as exc:
342            if raise_unregistered and isinstance(
343                exc, UnregisteredMessageIDError
344            ):
345                raise
346            return self._handle_raw_message_async_error(
347                bound_obj, msg_decoded, msgtype, exc
348            )
349
350        # Return an awaitable to handle the rest asynchronously.
351        return self._handle_raw_message_async(
352            bound_obj, msg_decoded, msgtype, handler_awaitable
353        )

Should be called when the receiver gets a message.

The return value is the raw response to the message.

class BoundMessageReceiver:
396class BoundMessageReceiver:
397    """Base bound receiver class."""
398
399    def __init__(
400        self,
401        obj: Any,
402        receiver: MessageReceiver,
403    ) -> None:
404        assert obj is not None
405        self._obj = obj
406        self._receiver = receiver
407
408    @property
409    def protocol(self) -> MessageProtocol:
410        """Protocol associated with this receiver."""
411        return self._receiver.protocol
412
413    def encode_error_response(self, exc: Exception) -> str:
414        """Given an error, return a response ready to send.
415
416        This should be used for any errors that happen outside of
417        standard handle_raw_message calls. Any errors within those
418        calls will be automatically returned as encoded strings.
419        """
420        # Passing None for Message here; we would only have that available
421        # for things going wrong in the handler (which this is not for).
422        return self._receiver.encode_error_response(self._obj, None, exc)[0]

Base bound receiver class.

BoundMessageReceiver(obj: Any, receiver: MessageReceiver)
399    def __init__(
400        self,
401        obj: Any,
402        receiver: MessageReceiver,
403    ) -> None:
404        assert obj is not None
405        self._obj = obj
406        self._receiver = receiver
protocol: MessageProtocol

Protocol associated with this receiver.

def encode_error_response(self, exc: Exception) -> str:
413    def encode_error_response(self, exc: Exception) -> str:
414        """Given an error, return a response ready to send.
415
416        This should be used for any errors that happen outside of
417        standard handle_raw_message calls. Any errors within those
418        calls will be automatically returned as encoded strings.
419        """
420        # Passing None for Message here; we would only have that available
421        # for things going wrong in the handler (which this is not for).
422        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