updated files

dist/ba_data/python/efro/message/__init__.py

@@ -0,0 +1,45 @@
+# Released under the MIT License. See LICENSE for details.
+#
+"""Functionality for sending and responding to messages.
+Supports static typing for message types and possible return types.
+"""
+
+from __future__ import annotations
+
+from efro.util import set_canonical_module
+from efro.message._protocol import MessageProtocol
+from efro.message._sender import MessageSender, BoundMessageSender
+from efro.message._receiver import MessageReceiver, BoundMessageReceiver
+from efro.message._module import create_sender_module, create_receiver_module
+from efro.message._message import (
+    Message,
+    Response,
+    SysResponse,
+    EmptySysResponse,
+    ErrorSysResponse,
+    StringResponse,
+    BoolResponse,
+    UnregisteredMessageIDError,
+)
+
+__all__ = [
+    'Message',
+    'Response',
+    'SysResponse',
+    'EmptySysResponse',
+    'ErrorSysResponse',
+    'StringResponse',
+    'BoolResponse',
+    'MessageProtocol',
+    'MessageSender',
+    'BoundMessageSender',
+    'MessageReceiver',
+    'BoundMessageReceiver',
+    'create_sender_module',
+    'create_receiver_module',
+    'UnregisteredMessageIDError',
+]
+
+# Have these things present themselves cleanly as 'thismodule.SomeClass'
+# instead of 'thismodule._internalmodule.SomeClass'
+set_canonical_module(module_globals=globals(), names=__all__)

dist/ba_data/python/efro/message/_message.py

@@ -0,0 +1,108 @@
+# Released under the MIT License. See LICENSE for details.
+#
+"""Functionality for sending and responding to messages.
+Supports static typing for message types and possible return types.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Annotated
+from dataclasses import dataclass
+from enum import Enum
+
+from efro.dataclassio import ioprepped, IOAttrs
+
+if TYPE_CHECKING:
+    pass
+
+
+class UnregisteredMessageIDError(Exception):
+    """A message or response id is not covered by our protocol."""
+
+
+class Message:
+    """Base class for messages."""
+
+    @classmethod
+    def get_response_types(cls) -> list[type[Response] | None]:
+        """Return all Response types this Message can return when sent.
+
+        The default implementation specifies a None return type.
+        """
+        return [None]
+
+
+class Response:
+    """Base class for responses to messages."""
+
+
+class SysResponse:
+    """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:
+        """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.
+        """
+        setattr(self, '_sr_local_exception', exc)
+
+    def get_local_exception(self) -> Exception | None:
+        """Fetch a local attached exception."""
+        value = getattr(self, '_sr_local_exception', None)
+        assert isinstance(value, Exception | None)
+        return value
+
+
+# Some standard response types:
+
+
+@ioprepped
+@dataclass
+class ErrorSysResponse(SysResponse):
+    """SysResponse saying some error has occurred for the send.
+
+    This generally results in an Exception being raised for the caller.
+    """
+
+    class ErrorType(Enum):
+        """Type of error that occurred while sending a message."""
+
+        REMOTE = 0
+        REMOTE_CLEAN = 1
+        LOCAL = 2
+        COMMUNICATION = 3
+        REMOTE_COMMUNICATION = 4
+
+    error_message: Annotated[str, IOAttrs('m')]
+    error_type: Annotated[ErrorType, IOAttrs('e')] = ErrorType.REMOTE
+
+
+@ioprepped
+@dataclass
+class EmptySysResponse(SysResponse):
+    """The response equivalent of None."""
+
+
+# TODO: could allow handlers to deal in raw values for these
+# types similar to how we allow None in place of EmptySysResponse.
+# Though not sure if they are widely used enough to warrant the
+# extra code complexity.
+@ioprepped
+@dataclass
+class BoolResponse(Response):
+    """A simple bool value response."""
+
+    value: Annotated[bool, IOAttrs('v')]
+
+
+@ioprepped
+@dataclass
+class StringResponse(Response):
+    """A simple string value response."""
+
+    value: Annotated[str, IOAttrs('v')]

dist/ba_data/python/efro/message/_module.py

@@ -0,0 +1,109 @@
+# Released under the MIT License. See LICENSE for details.
+#
+"""Functionality for sending and responding to messages.
+Supports static typing for message types and possible return types.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from efro.message._protocol import MessageProtocol
+
+if TYPE_CHECKING:
+    pass
+
+
+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:
+    """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.
+    """
+    protocol = _protocol_from_code(
+        build_time_protocol_create_code
+        if build_time_protocol_create_code is not None
+        else protocol_create_code
+    )
+    return protocol.do_create_sender_module(
+        basename=basename,
+        protocol_create_code=protocol_create_code,
+        enable_sync_sends=enable_sync_sends,
+        enable_async_sends=enable_async_sends,
+        private=private,
+        protocol_module_level_import_code=protocol_module_level_import_code,
+    )
+
+
+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:
+    """ "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.
+    """
+    protocol = _protocol_from_code(
+        build_time_protocol_create_code
+        if build_time_protocol_create_code is not None
+        else protocol_create_code
+    )
+    return protocol.do_create_receiver_module(
+        basename=basename,
+        protocol_create_code=protocol_create_code,
+        is_async=is_async,
+        private=private,
+        protocol_module_level_import_code=protocol_module_level_import_code,
+    )
+
+
+def _protocol_from_code(protocol_create_code: str) -> MessageProtocol:
+    env: dict = {}
+    exec(protocol_create_code, env)  # pylint: disable=exec-used
+    protocol = env.get('protocol')
+    if not isinstance(protocol, MessageProtocol):
+        raise RuntimeError(
+            f'protocol_create_code yielded'
+            f' a {type(protocol)}; expected a MessageProtocol instance.'
+        )
+    return protocol

dist/ba_data/python/efro/message/_protocol.py

@@ -0,0 +1,653 @@
+# Released under the MIT License. See LICENSE for details.
+#
+"""Functionality for sending and responding to messages.
+Supports static typing for message types and possible return types.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+import traceback
+import json
+
+from efro.error import CleanError, CommunicationError
+from efro.dataclassio import (
+    is_ioprepped_dataclass,
+    dataclass_to_dict,
+    dataclass_from_dict,
+)
+from efro.message._message import (
+    Message,
+    Response,
+    SysResponse,
+    ErrorSysResponse,
+    EmptySysResponse,
+    UnregisteredMessageIDError,
+)
+
+if TYPE_CHECKING:
+    from typing import Any, Literal
+
+
+class MessageProtocol:
+    """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.
+    """
+
+    def __init__(
+        self,
+        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,
+    ) -> None:
+        """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.
+        """
+        # pylint: disable=too-many-locals
+        self.message_types_by_id: dict[int, type[Message]] = {}
+        self.message_ids_by_type: dict[type[Message], int] = {}
+        self.response_types_by_id: dict[
+            int, type[Response] | type[SysResponse]
+        ] = {}
+        self.response_ids_by_type: dict[
+            type[Response] | type[SysResponse], int
+        ] = {}
+        for m_id, m_type in message_types.items():
+
+            # Make sure only valid message types were passed and each
+            # id was assigned only once.
+            assert isinstance(m_id, int)
+            assert m_id >= 0
+            assert is_ioprepped_dataclass(m_type) and issubclass(
+                m_type, Message
+            )
+            assert self.message_types_by_id.get(m_id) is None
+            self.message_types_by_id[m_id] = m_type
+            self.message_ids_by_type[m_type] = m_id
+
+        for r_id, r_type in response_types.items():
+            assert isinstance(r_id, int)
+            assert r_id >= 0
+            assert is_ioprepped_dataclass(r_type) and issubclass(
+                r_type, Response
+            )
+            assert self.response_types_by_id.get(r_id) is None
+            self.response_types_by_id[r_id] = r_type
+            self.response_ids_by_type[r_type] = r_id
+
+        # Register our SysResponse types. These use negative
+        # IDs so as to never overlap with user Response types.
+        def _reg_sys(reg_tp: type[SysResponse], reg_id: int) -> None:
+            assert self.response_types_by_id.get(reg_id) is None
+            self.response_types_by_id[reg_id] = reg_tp
+            self.response_ids_by_type[reg_tp] = reg_id
+
+        _reg_sys(ErrorSysResponse, -1)
+        _reg_sys(EmptySysResponse, -2)
+
+        # Some extra-thorough validation in debug mode.
+        if __debug__:
+            # Make sure all Message types' return types are valid
+            # and have been assigned an ID as well.
+            all_response_types: set[type[Response] | None] = set()
+            for m_id, m_type in message_types.items():
+                m_rtypes = m_type.get_response_types()
+
+                assert isinstance(m_rtypes, list)
+                assert (
+                    m_rtypes
+                ), f'Message type {m_type} specifies no return types.'
+                assert len(set(m_rtypes)) == len(m_rtypes)  # check dups
+                for m_rtype in m_rtypes:
+                    all_response_types.add(m_rtype)
+            for cls in all_response_types:
+                if cls is None:
+                    continue
+                assert is_ioprepped_dataclass(cls)
+                assert issubclass(cls, Response)
+                if cls not in self.response_ids_by_type:
+                    raise ValueError(
+                        f'Possible response type {cls} needs to be included'
+                        f' in response_types for this protocol.'
+                    )
+
+            # Make sure all registered types have unique base names.
+            # We can take advantage of this to generate cleaner looking
+            # protocol modules. Can revisit if this is ever a problem.
+            mtypenames = set(tp.__name__ for tp in self.message_ids_by_type)
+            if len(mtypenames) != len(message_types):
+                raise ValueError(
+                    'message_types contains duplicate __name__s;'
+                    ' all types are required to have unique names.'
+                )
+
+        self.forward_clean_errors = forward_clean_errors
+        self.forward_communication_errors = forward_communication_errors
+        self.remote_errors_include_stack_traces = (
+            remote_errors_include_stack_traces
+        )
+        self.log_remote_errors = log_remote_errors
+
+    @staticmethod
+    def encode_dict(obj: dict) -> str:
+        """Json-encode a provided dict."""
+        return json.dumps(obj, separators=(',', ':'))
+
+    def message_to_dict(self, message: Message) -> dict:
+        """Encode a message to a json ready dict."""
+        return self._to_dict(message, self.message_ids_by_type, 'message')
+
+    def response_to_dict(self, response: Response | SysResponse) -> dict:
+        """Encode a response to a json ready dict."""
+        return self._to_dict(response, self.response_ids_by_type, 'response')
+
+    def error_to_response(self, exc: Exception) -> tuple[SysResponse, bool]:
+        """Translate an Exception to a SysResponse.
+
+        Also returns whether the error should be logged if this happened
+        within handle_raw_message().
+        """
+
+        # If anything goes wrong, return a ErrorSysResponse instead.
+        # (either CLEAN or generic REMOTE)
+        if self.forward_clean_errors and isinstance(exc, CleanError):
+            return (
+                ErrorSysResponse(
+                    error_message=str(exc),
+                    error_type=ErrorSysResponse.ErrorType.REMOTE_CLEAN,
+                ),
+                False,
+            )
+        if self.forward_communication_errors and isinstance(
+            exc, CommunicationError
+        ):
+            return (
+                ErrorSysResponse(
+                    error_message=str(exc),
+                    error_type=ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION,
+                ),
+                False,
+            )
+        return (
+            ErrorSysResponse(
+                error_message=(
+                    traceback.format_exc()
+                    if self.remote_errors_include_stack_traces
+                    else 'An internal error has occurred.'
+                ),
+                error_type=ErrorSysResponse.ErrorType.REMOTE,
+            ),
+            self.log_remote_errors,
+        )
+
+    def _to_dict(
+        self, message: Any, ids_by_type: dict[type, int], opname: str
+    ) -> dict:
+        """Encode a message to a json string for transport."""
+
+        m_id: int | None = ids_by_type.get(type(message))
+        if m_id is None:
+            raise TypeError(
+                f'{opname} type is not registered in protocol:'
+                f' {type(message)}'
+            )
+        out = {'t': m_id, 'm': dataclass_to_dict(message)}
+        return out
+
+    @staticmethod
+    def decode_dict(data: str) -> dict:
+        """Decode data to a dict."""
+        out = json.loads(data)
+        assert isinstance(out, dict)
+        return out
+
+    def message_from_dict(self, data: dict) -> Message:
+        """Decode a message from a json string."""
+        out = self._from_dict(data, self.message_types_by_id, 'message')
+        assert isinstance(out, Message)
+        return out
+
+    def response_from_dict(self, data: dict) -> Response | SysResponse:
+        """Decode a response from a json string."""
+        out = self._from_dict(data, self.response_types_by_id, 'response')
+        assert isinstance(out, Response | SysResponse)
+        return out
+
+    # Weeeird; we get mypy errors returning dict[int, type] but
+    # dict[int, typing.Type] or dict[int, type[Any]] works..
+    def _from_dict(
+        self, data: dict, types_by_id: dict[int, type[Any]], opname: str
+    ) -> Any:
+        """Decode a message from a json string."""
+        msgdict: dict | None
+
+        m_id = data.get('t')
+        # Allow omitting 'm' dict if its empty.
+        msgdict = data.get('m', {})
+
+        assert isinstance(m_id, int)
+        assert isinstance(msgdict, dict)
+
+        # Decode this particular type.
+        msgtype = types_by_id.get(m_id)
+        if msgtype is None:
+            raise UnregisteredMessageIDError(
+                f'Got unregistered {opname} id of {m_id}.'
+            )
+        return dataclass_from_dict(msgtype, msgdict)
+
+    def _get_module_header(
+        self,
+        part: Literal['sender', 'receiver'],
+        extra_import_code: str | None,
+        enable_async_sends: bool,
+    ) -> str:
+        """Return common parts of generated modules."""
+        # pylint: disable=too-many-locals
+        # pylint: disable=too-many-branches
+        # pylint: disable=too-many-statements
+        import textwrap
+
+        tpimports: dict[str, list[str]] = {}
+        imports: dict[str, list[str]] = {}
+
+        single_message_type = len(self.message_ids_by_type) == 1
+
+        msgtypes = list(self.message_ids_by_type)
+        if part == 'sender':
+            msgtypes.append(Message)
+        for msgtype in msgtypes:
+            tpimports.setdefault(msgtype.__module__, []).append(
+                msgtype.__name__
+            )
+        rsptypes = list(self.response_ids_by_type)
+        if part == 'sender':
+            rsptypes.append(Response)
+        for rsp_tp in rsptypes:
+            # Skip these as they don't actually show up in code.
+            if rsp_tp is EmptySysResponse or rsp_tp is ErrorSysResponse:
+                continue
+            if (
+                single_message_type
+                and part == 'sender'
+                and rsp_tp is not Response
+            ):
+                # We need to cast to the single supported response type
+                # in this case so need response types at runtime.
+                imports.setdefault(rsp_tp.__module__, []).append(
+                    rsp_tp.__name__
+                )
+            else:
+                tpimports.setdefault(rsp_tp.__module__, []).append(
+                    rsp_tp.__name__
+                )
+
+        import_lines = ''
+        tpimport_lines = ''
+
+        for module, names in sorted(imports.items()):
+            jnames = ', '.join(names)
+            line = f'from {module} import {jnames}'
+            if len(line) > 79:
+                # Recreate in a wrapping-friendly form.
+                line = f'from {module} import ({jnames})'
+            import_lines += f'{line}\n'
+        for module, names in sorted(tpimports.items()):
+            jnames = ', '.join(names)
+            line = f'from {module} import {jnames}'
+            if len(line) > 75:  # Account for indent
+                # Recreate in a wrapping-friendly form.
+                line = f'from {module} import ({jnames})'
+            tpimport_lines += f'{line}\n'
+
+        if part == 'sender':
+            import_lines += (
+                'from efro.message import MessageSender, BoundMessageSender'
+            )
+            tpimport_typing_extras = ''
+        else:
+            if single_message_type:
+                import_lines += (
+                    'from efro.message import (MessageReceiver,'
+                    ' BoundMessageReceiver, Message, Response)'
+                )
+            else:
+                import_lines += (
+                    'from efro.message import MessageReceiver,'
+                    ' BoundMessageReceiver'
+                )
+            tpimport_typing_extras = ', Awaitable'
+
+        if extra_import_code is not None:
+            import_lines += f'\n{extra_import_code}\n'
+
+        ovld = ', overload' if not single_message_type else ''
+        ovld2 = (
+            ', cast, Awaitable'
+            if (single_message_type and part == 'sender' and enable_async_sends)
+            else ''
+        )
+        tpimport_lines = textwrap.indent(tpimport_lines, '    ')
+
+        baseimps = ['Any']
+        if part == 'receiver':
+            baseimps.append('Callable')
+        if part == 'sender' and enable_async_sends:
+            baseimps.append('Awaitable')
+        baseimps_s = ', '.join(baseimps)
+        out = (
+            '# Released under the MIT License. See LICENSE for details.\n'
+            f'#\n'
+            f'"""Auto-generated {part} module. Do not edit by hand."""\n'
+            f'\n'
+            f'from __future__ import annotations\n'
+            f'\n'
+            f'from typing import TYPE_CHECKING{ovld}{ovld2}\n'
+            f'\n'
+            f'{import_lines}\n'
+            f'\n'
+            f'if TYPE_CHECKING:\n'
+            f'    from typing import {baseimps_s}'
+            f'{tpimport_typing_extras}\n'
+            f'{tpimport_lines}'
+            f'\n'
+            f'\n'
+        )
+        return out
+
+    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:
+        """Used by create_sender_module(); do not call directly."""
+        # pylint: disable=too-many-locals
+        # pylint: disable=too-many-branches
+        import textwrap
+
+        msgtypes = list(self.message_ids_by_type.keys())
+
+        ppre = '_' if private else ''
+        out = self._get_module_header(
+            'sender',
+            extra_import_code=protocol_module_level_import_code,
+            enable_async_sends=enable_async_sends,
+        )
+        ccind = textwrap.indent(protocol_create_code, '        ')
+        out += (
+            f'class {ppre}{basename}(MessageSender):\n'
+            f'    """Protocol-specific sender."""\n'
+            f'\n'
+            f'    def __init__(self) -> None:\n'
+            f'{ccind}\n'
+            f'        super().__init__(protocol)\n'
+            f'\n'
+            f'    def __get__(\n'
+            f'        self, obj: Any, type_in: Any = None\n'
+            f'    ) -> {ppre}Bound{basename}:\n'
+            f'        return {ppre}Bound{basename}(obj, self)\n'
+            f'\n'
+            f'\n'
+            f'class {ppre}Bound{basename}(BoundMessageSender):\n'
+            f'    """Protocol-specific bound sender."""\n'
+        )
+
+        def _filt_tp_name(rtype: type[Response] | None) -> str:
+            return 'None' if rtype is None else rtype.__name__
+
+        # Define handler() overloads for all registered message types.
+        if msgtypes:
+            for async_pass in False, True:
+                if async_pass and not enable_async_sends:
+                    continue
+                if not async_pass and not enable_sync_sends:
+                    continue
+                pfx = 'async ' if async_pass else ''
+                sfx = '_async' if async_pass else ''
+                # awt = 'await ' if async_pass else ''
+                awt = ''
+                how = 'asynchronously' if async_pass else 'synchronously'
+
+                if len(msgtypes) == 1:
+                    # Special case: with a single message types we don't
+                    # use overloads.
+                    msgtype = msgtypes[0]
+                    msgtypevar = msgtype.__name__
+                    rtypes = msgtype.get_response_types()
+                    if len(rtypes) > 1:
+                        rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
+                    else:
+                        rtypevar = _filt_tp_name(rtypes[0])
+                    if async_pass:
+                        rtypevar = f'Awaitable[{rtypevar}]'
+                    out += (
+                        f'\n'
+                        f'    def send{sfx}(self,'
+                        f' message: {msgtypevar})'
+                        f' -> {rtypevar}:\n'
+                        f'        """Send a message {how}."""\n'
+                        f'        out = {awt}self._sender.'
+                        f'send{sfx}(self._obj, message)\n'
+                    )
+                    if not async_pass:
+                        out += (
+                            f'        assert isinstance(out, {rtypevar})\n'
+                            '        return out\n'
+                        )
+                    else:
+                        out += f'        return cast({rtypevar}, out)\n'
+
+                else:
+
+                    for msgtype in msgtypes:
+                        msgtypevar = msgtype.__name__
+                        rtypes = msgtype.get_response_types()
+                        if len(rtypes) > 1:
+                            rtypevar = ' | '.join(
+                                _filt_tp_name(t) for t in rtypes
+                            )
+                        else:
+                            rtypevar = _filt_tp_name(rtypes[0])
+                        out += (
+                            f'\n'
+                            f'    @overload\n'
+                            f'    {pfx}def send{sfx}(self,'
+                            f' message: {msgtypevar})'
+                            f' -> {rtypevar}:\n'
+                            f'        ...\n'
+                        )
+                    rtypevar = 'Response | None'
+                    if async_pass:
+                        rtypevar = f'Awaitable[{rtypevar}]'
+                    out += (
+                        f'\n'
+                        f'    def send{sfx}(self, message: Message)'
+                        f' -> {rtypevar}:\n'
+                        f'        """Send a message {how}."""\n'
+                        f'        return {awt}self._sender.'
+                        f'send{sfx}(self._obj, message)\n'
+                    )
+
+        return out
+
+    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:
+        """Used by create_receiver_module(); do not call directly."""
+        # pylint: disable=too-many-locals
+        import textwrap
+
+        desc = 'asynchronous' if is_async else 'synchronous'
+        ppre = '_' if private else ''
+        msgtypes = list(self.message_ids_by_type.keys())
+        out = self._get_module_header(
+            'receiver',
+            extra_import_code=protocol_module_level_import_code,
+            enable_async_sends=False,
+        )
+        ccind = textwrap.indent(protocol_create_code, '        ')
+        out += (
+            f'class {ppre}{basename}(MessageReceiver):\n'
+            f'    """Protocol-specific {desc} receiver."""\n'
+            f'\n'
+            f'    is_async = {is_async}\n'
+            f'\n'
+            f'    def __init__(self) -> None:\n'
+            f'{ccind}\n'
+            f'        super().__init__(protocol)\n'
+            f'\n'
+            f'    def __get__(\n'
+            f'        self,\n'
+            f'        obj: Any,\n'
+            f'        type_in: Any = None,\n'
+            f'    ) -> {ppre}Bound{basename}:\n'
+            f'        return {ppre}Bound{basename}('
+            f'obj, self)\n'
+        )
+
+        # Define handler() overloads for all registered message types.
+
+        def _filt_tp_name(rtype: type[Response] | None) -> str:
+            return 'None' if rtype is None else rtype.__name__
+
+        if msgtypes:
+            cbgn = 'Awaitable[' if is_async else ''
+            cend = ']' if is_async else ''
+            if len(msgtypes) == 1:
+                # Special case: when we have a single message type we don't
+                # use overloads.
+                msgtype = msgtypes[0]
+                msgtypevar = msgtype.__name__
+                rtypes = msgtype.get_response_types()
+                if len(rtypes) > 1:
+                    rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
+                else:
+                    rtypevar = _filt_tp_name(rtypes[0])
+                rtypevar = f'{cbgn}{rtypevar}{cend}'
+                out += (
+                    f'\n'
+                    f'    def handler(\n'
+                    f'        self,\n'
+                    f'        call: Callable[[Any, {msgtypevar}], '
+                    f'{rtypevar}],\n'
+                    f'    )'
+                    f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n'
+                    f'        """Decorator to register message handlers."""\n'
+                    f'        from typing import cast, Callable, Any\n'
+                    f'\n'
+                    f'        self.register_handler(cast(Callable'
+                    f'[[Any, Message], Response], call))\n'
+                    f'        return call\n'
+                )
+            else:
+                for msgtype in msgtypes:
+                    msgtypevar = msgtype.__name__
+                    rtypes = msgtype.get_response_types()
+                    if len(rtypes) > 1:
+                        rtypevar = ' | '.join(_filt_tp_name(t) for t in rtypes)
+                    else:
+                        rtypevar = _filt_tp_name(rtypes[0])
+                    rtypevar = f'{cbgn}{rtypevar}{cend}'
+                    out += (
+                        f'\n'
+                        f'    @overload\n'
+                        f'    def handler(\n'
+                        f'        self,\n'
+                        f'        call: Callable[[Any, {msgtypevar}], '
+                        f'{rtypevar}],\n'
+                        f'    )'
+                        f' -> Callable[[Any, {msgtypevar}], {rtypevar}]:\n'
+                        f'        ...\n'
+                    )
+                out += (
+                    '\n'
+                    '    def handler(self, call: Callable) -> Callable:\n'
+                    '        """Decorator to register message handlers."""\n'
+                    '        self.register_handler(call)\n'
+                    '        return call\n'
+                )
+
+        out += (
+            f'\n'
+            f'\n'
+            f'class {ppre}Bound{basename}(BoundMessageReceiver):\n'
+            f'    """Protocol-specific bound receiver."""\n'
+        )
+        if is_async:
+            out += (
+                '\n'
+                '    def handle_raw_message(\n'
+                '        self, message: str, raise_unregistered: bool = False\n'
+                '    ) -> Awaitable[str]:\n'
+                '        """Asynchronously handle a raw incoming message."""\n'
+                '        return self._receiver.'
+                'handle_raw_message_async(\n'
+                '            self._obj, message, raise_unregistered\n'
+                '        )\n'
+            )
+
+        else:
+            out += (
+                '\n'
+                '    def handle_raw_message(\n'
+                '        self, message: str, raise_unregistered: bool = False\n'
+                '    ) -> str:\n'
+                '        """Synchronously handle a raw incoming message."""\n'
+                '        return self._receiver.handle_raw_message(\n'
+                '            self._obj, message, raise_unregistered\n'
+                '        )\n'
+            )
+
+        return out

dist/ba_data/python/efro/message/_receiver.py

@@ -0,0 +1,420 @@
+# Released under the MIT License. See LICENSE for details.
+#
+"""Functionality for sending and responding to messages.
+Supports static typing for message types and possible return types.
+"""
+
+from __future__ import annotations
+
+import types
+import inspect
+import logging
+from typing import TYPE_CHECKING
+
+from efro.message._message import (
+    Message,
+    Response,
+    EmptySysResponse,
+    UnregisteredMessageIDError,
+)
+
+if TYPE_CHECKING:
+    from typing import Any, Callable, Awaitable
+
+    from efro.message._protocol import MessageProtocol
+    from efro.message._message import SysResponse
+
+
+class MessageReceiver:
+    """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.
+    """
+
+    is_async = False
+
+    def __init__(self, protocol: MessageProtocol) -> None:
+        self.protocol = protocol
+        self._handlers: dict[type[Message], Callable] = {}
+        self._decode_filter_call: Callable[
+            [Any, dict, Message], None
+        ] | None = None
+        self._encode_filter_call: Callable[
+            [Any, Message | None, Response | SysResponse, dict], None
+        ] | None = None
+
+    # noinspection PyProtectedMember
+    def register_handler(
+        self, call: Callable[[Any, Message], Response | None]
+    ) -> None:
+        """Register a handler call.
+
+        The message type handled by the call is determined by its
+        type annotation.
+        """
+        # TODO: can use types.GenericAlias in 3.9.
+        # (hmm though now that we're there,  it seems a drop-in
+        # replace gives us errors. Should re-test in 3.10 as it seems
+        # that typing_extensions handles it differently in that case)
+        from typing import _GenericAlias  # type: ignore
+        from typing import get_type_hints, get_args
+
+        sig = inspect.getfullargspec(call)
+
+        # The provided callable should be a method taking one 'msg' arg.
+        expectedsig = ['self', 'msg']
+        if sig.args != expectedsig:
+            raise ValueError(
+                f'Expected callable signature of {expectedsig};'
+                f' got {sig.args}'
+            )
+
+        # Make sure we are only given async methods if we are an async handler
+        # and sync ones otherwise.
+        # UPDATE - can't do this anymore since we now sometimes use
+        # regular functions which return awaitables instead of having
+        # the entire function be async.
+        # is_async = inspect.iscoroutinefunction(call)
+        # if self.is_async != is_async:
+        #     msg = (
+        #         'Expected a sync method; found an async one.'
+        #         if is_async
+        #         else 'Expected an async method; found a sync one.'
+        #     )
+        #     raise ValueError(msg)
+
+        # Check annotation types to determine what message types we handle.
+        # Return-type annotation can be a Union, but we probably don't
+        # have it available at runtime. Explicitly pull it in.
+        # UPDATE: we've updated our pylint filter to where we should
+        # have all annotations available.
+        # anns = get_type_hints(call, localns={'Union': Union})
+        anns = get_type_hints(call)
+
+        msgtype = anns.get('msg')
+        if not isinstance(msgtype, type):
+            raise TypeError(
+                f'expected a type for "msg" annotation; got {type(msgtype)}.'
+            )
+        assert issubclass(msgtype, Message)
+
+        ret = anns.get('return')
+        responsetypes: tuple[type[Any] | None, ...]
+
+        # Return types can be a single type or a union of types.
+        if isinstance(ret, (_GenericAlias, types.UnionType)):
+            targs = get_args(ret)
+            if not all(isinstance(a, (type, type(None))) for a in targs):
+                raise TypeError(
+                    f'expected only types for "return" annotation;'
+                    f' got {targs}.'
+                )
+            responsetypes = targs
+        else:
+            if not isinstance(ret, (type, type(None))):
+                raise TypeError(
+                    f'expected one or more types for'
+                    f' "return" annotation; got a {type(ret)}.'
+                )
+            # This seems like maybe a mypy bug. Appeared after adding
+            # types.UnionType above.
+            responsetypes = (ret,)
+
+        # This will contain NoneType for empty return cases, but
+        # we expect it to be None.
+        responsetypes = tuple(
+            None if r is type(None) else r for r in responsetypes
+        )
+
+        # Make sure our protocol has this message type registered and our
+        # return types exactly match. (Technically we could return a subset
+        # of the supported types; can allow this in the future if it makes
+        # sense).
+        registered_types = self.protocol.message_ids_by_type.keys()
+
+        if msgtype not in registered_types:
+            raise TypeError(
+                f'Message type {msgtype} is not registered'
+                f' in this Protocol.'
+            )
+
+        if msgtype in self._handlers:
+            raise TypeError(
+                f'Message type {msgtype} already has a registered' f' handler.'
+            )
+
+        # Make sure the responses exactly matches what the message expects.
+        if set(responsetypes) != set(msgtype.get_response_types()):
+            raise TypeError(
+                f'Provided response types {responsetypes} do not'
+                f' match the set expected by message type {msgtype}: '
+                f'({msgtype.get_response_types()})'
+            )
+
+        # Ok; we're good!
+        self._handlers[msgtype] = call
+
+    def decode_filter_method(
+        self, call: Callable[[Any, dict, Message], None]
+    ) -> Callable[[Any, dict, Message], None]:
+        """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()
+        """
+        assert self._decode_filter_call is None
+        self._decode_filter_call = call
+        return call
+
+    def encode_filter_method(
+        self,
+        call: Callable[
+            [Any, Message | None, Response | SysResponse, dict], None
+        ],
+    ) -> Callable[[Any, Message | None, Response, dict], None]:
+        """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.
+        """
+        assert self._encode_filter_call is None
+        self._encode_filter_call = call
+        return call
+
+    def validate(self, log_only: bool = False) -> None:
+        """Check for handler completeness, valid types, etc."""
+        for msgtype in self.protocol.message_ids_by_type.keys():
+            if issubclass(msgtype, Response):
+                continue
+            if msgtype not in self._handlers:
+                msg = (
+                    f'Protocol message type {msgtype} is not handled'
+                    f' by receiver type {type(self)}.'
+                )
+                if log_only:
+                    logging.error(msg)
+                else:
+                    raise TypeError(msg)
+
+    def _decode_incoming_message_base(
+        self, bound_obj: Any, msg: str
+    ) -> tuple[Any, dict, Message]:
+        # Decode the incoming message.
+        msg_dict = self.protocol.decode_dict(msg)
+        msg_decoded = self.protocol.message_from_dict(msg_dict)
+        assert isinstance(msg_decoded, Message)
+        if self._decode_filter_call is not None:
+            self._decode_filter_call(bound_obj, msg_dict, msg_decoded)
+        return bound_obj, msg_dict, msg_decoded
+
+    def _decode_incoming_message(self, bound_obj: Any, msg: str) -> Message:
+        bound_obj, _msg_dict, msg_decoded = self._decode_incoming_message_base(
+            bound_obj=bound_obj, msg=msg
+        )
+        return msg_decoded
+
+    def encode_user_response(
+        self, bound_obj: Any, message: Message, response: Response | None
+    ) -> str:
+        """Encode a response provided by the user for sending."""
+
+        assert isinstance(response, Response | None)
+        # (user should never explicitly return error-responses)
+        assert (
+            response is None or type(response) in message.get_response_types()
+        )
+
+        # A return value of None equals EmptySysResponse.
+        out_response: Response | SysResponse
+        if response is None:
+            out_response = EmptySysResponse()
+        else:
+            out_response = response
+
+        response_dict = self.protocol.response_to_dict(out_response)
+        if self._encode_filter_call is not None:
+            self._encode_filter_call(
+                bound_obj, message, out_response, response_dict
+            )
+        return self.protocol.encode_dict(response_dict)
+
+    def encode_error_response(
+        self, bound_obj: Any, message: Message | None, exc: Exception
+    ) -> tuple[str, bool]:
+        """Given an error, return sysresponse str and whether to log."""
+        response, dolog = self.protocol.error_to_response(exc)
+        response_dict = self.protocol.response_to_dict(response)
+        if self._encode_filter_call is not None:
+            self._encode_filter_call(
+                bound_obj, message, response, response_dict
+            )
+        return self.protocol.encode_dict(response_dict), dolog
+
+    def handle_raw_message(
+        self, bound_obj: Any, msg: str, raise_unregistered: bool = False
+    ) -> str:
+        """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.
+        """
+        assert not self.is_async, "can't call sync handler on async receiver"
+        msg_decoded: Message | None = None
+        msgtype: type[Message] | None = None
+        try:
+            msg_decoded = self._decode_incoming_message(bound_obj, msg)
+            msgtype = type(msg_decoded)
+            handler = self._handlers.get(msgtype)
+            if handler is None:
+                raise RuntimeError(f'Got unhandled message type: {msgtype}.')
+            response = handler(bound_obj, msg_decoded)
+            assert isinstance(response, Response | None)
+            return self.encode_user_response(bound_obj, msg_decoded, response)
+
+        except Exception as exc:
+            if raise_unregistered and isinstance(
+                exc, UnregisteredMessageIDError
+            ):
+                raise
+            rstr, dolog = self.encode_error_response(
+                bound_obj, msg_decoded, exc
+            )
+            if dolog:
+                if msgtype is not None:
+                    logging.exception(
+                        'Error handling %s.%s message.',
+                        msgtype.__module__,
+                        msgtype.__qualname__,
+                    )
+                else:
+                    logging.exception('Error in efro.message handling.')
+            return rstr
+
+    def handle_raw_message_async(
+        self, bound_obj: Any, msg: str, raise_unregistered: bool = False
+    ) -> Awaitable[str]:
+        """Should be called when the receiver gets a message.
+
+        The return value is the raw response to the message.
+        """
+
+        # Note: This call is synchronous so that the first part of it can
+        # happen synchronously. If the whole call were async we wouldn't be
+        # able to guarantee that messages handlers would be called in the
+        # order the messages were received.
+
+        assert self.is_async, "can't call async handler on sync receiver"
+        msg_decoded: Message | None = None
+        msgtype: type[Message] | None = None
+        try:
+            msg_decoded = self._decode_incoming_message(bound_obj, msg)
+            msgtype = type(msg_decoded)
+            handler = self._handlers.get(msgtype)
+            if handler is None:
+                raise RuntimeError(f'Got unhandled message type: {msgtype}.')
+            handler_awaitable = handler(bound_obj, msg_decoded)
+
+        except Exception as exc:
+            if raise_unregistered and isinstance(
+                exc, UnregisteredMessageIDError
+            ):
+                raise
+            return self._handle_raw_message_async_error(
+                bound_obj, msg_decoded, msgtype, exc
+            )
+
+        # Return an awaitable to handle the rest asynchronously.
+        return self._handle_raw_message_async(
+            bound_obj, msg_decoded, msgtype, handler_awaitable
+        )
+
+    async def _handle_raw_message_async_error(
+        self,
+        bound_obj: Any,
+        msg_decoded: Message | None,
+        msgtype: type[Message] | None,
+        exc: Exception,
+    ) -> str:
+        rstr, dolog = self.encode_error_response(bound_obj, msg_decoded, exc)
+        if dolog:
+            if msgtype is not None:
+                logging.exception(
+                    'Error handling %s.%s message.',
+                    msgtype.__module__,
+                    msgtype.__qualname__,
+                )
+            else:
+                logging.exception('Error in efro.message handling.')
+        return rstr
+
+    async def _handle_raw_message_async(
+        self,
+        bound_obj: Any,
+        msg_decoded: Message,
+        msgtype: type[Message] | None,
+        handler_awaitable: Awaitable[Response | None],
+    ) -> str:
+        """Should be called when the receiver gets a message.
+
+        The return value is the raw response to the message.
+        """
+        try:
+            response = await handler_awaitable
+            assert isinstance(response, Response | None)
+            return self.encode_user_response(bound_obj, msg_decoded, response)
+
+        except Exception as exc:
+            return await self._handle_raw_message_async_error(
+                bound_obj, msg_decoded, msgtype, exc
+            )
+
+
+class BoundMessageReceiver:
+    """Base bound receiver class."""
+
+    def __init__(
+        self,
+        obj: Any,
+        receiver: MessageReceiver,
+    ) -> None:
+        assert obj is not None
+        self._obj = obj
+        self._receiver = receiver
+
+    @property
+    def protocol(self) -> MessageProtocol:
+        """Protocol associated with this receiver."""
+        return self._receiver.protocol
+
+    def encode_error_response(self, exc: Exception) -> str:
+        """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.
+        """
+        # Passing None for Message here; we would only have that available
+        # for things going wrong in the handler (which this is not for).
+        return self._receiver.encode_error_response(self._obj, None, exc)[0]

dist/ba_data/python/efro/message/_sender.py

@@ -0,0 +1,465 @@
+# Released under the MIT License. See LICENSE for details.
+#
+"""Functionality for sending and responding to messages.
+Supports static typing for message types and possible return types.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from efro.error import CleanError, RemoteError, CommunicationError
+from efro.message._message import EmptySysResponse, ErrorSysResponse, Response
+
+if TYPE_CHECKING:
+    from typing import Any, Callable, Awaitable
+
+    from efro.message._message import Message, SysResponse
+    from efro.message._protocol import MessageProtocol
+
+
+class MessageSender:
+    """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())
+    """
+
+    def __init__(self, protocol: MessageProtocol) -> None:
+        self.protocol = protocol
+        self._send_raw_message_call: Callable[[Any, str], str] | None = None
+        self._send_async_raw_message_call: Callable[
+            [Any, str], Awaitable[str]
+        ] | None = None
+        self._send_async_raw_message_ex_call: Callable[
+            [Any, str, Message], Awaitable[str]
+        ] | None = None
+        self._encode_filter_call: Callable[
+            [Any, Message, dict], None
+        ] | None = None
+        self._decode_filter_call: Callable[
+            [Any, Message, dict, Response | SysResponse], None
+        ] | None = None
+        self._peer_desc_call: Callable[[Any], str] | None = None
+
+    def send_method(
+        self, call: Callable[[Any, str], str]
+    ) -> Callable[[Any, str], str]:
+        """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.
+        """
+        assert self._send_raw_message_call is None
+        self._send_raw_message_call = call
+        return call
+
+    def send_async_method(
+        self, call: Callable[[Any, str], Awaitable[str]]
+    ) -> Callable[[Any, str], Awaitable[str]]:
+        """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.
+        """
+        assert self._send_async_raw_message_call is None
+        self._send_async_raw_message_call = call
+        return call
+
+    def send_async_ex_method(
+        self, call: Callable[[Any, str, Message], Awaitable[str]]
+    ) -> Callable[[Any, str, Message], Awaitable[str]]:
+        """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.
+        """
+        assert self._send_async_raw_message_ex_call is None
+        self._send_async_raw_message_ex_call = call
+        return call
+
+    def encode_filter_method(
+        self, call: Callable[[Any, Message, dict], None]
+    ) -> Callable[[Any, Message, dict], None]:
+        """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.
+        """
+        assert self._encode_filter_call is None
+        self._encode_filter_call = call
+        return call
+
+    def decode_filter_method(
+        self, call: Callable[[Any, Message, dict, Response | SysResponse], None]
+    ) -> Callable[[Any, Message, dict, Response], None]:
+        """Function decorator for defining a decode filter.
+
+        Decode filters can be used to extract extra data from incoming
+        message dicts.
+        """
+        assert self._decode_filter_call is None
+        self._decode_filter_call = call
+        return call
+
+    def peer_desc_method(
+        self, call: Callable[[Any], str]
+    ) -> Callable[[Any], str]:
+        """Function decorator for defining peer descriptions.
+
+        These are included in error messages or other diagnostics.
+        """
+        assert self._peer_desc_call is None
+        self._peer_desc_call = call
+        return call
+
+    def send(self, bound_obj: Any, message: Message) -> Response | None:
+        """Send a message synchronously."""
+        return self.unpack_raw_response(
+            bound_obj=bound_obj,
+            message=message,
+            raw_response=self.fetch_raw_response(
+                bound_obj=bound_obj,
+                message=message,
+            ),
+        )
+
+    def send_async(
+        self, bound_obj: Any, message: Message
+    ) -> Awaitable[Response | None]:
+        """Send a message asynchronously."""
+
+        # Note: This call is synchronous so that the first part of it can
+        # happen synchronously. If the whole call were async we wouldn't be
+        # able to guarantee that messages sent in order would actually go
+        # out in order.
+        raw_response_awaitable = self.fetch_raw_response_async(
+            bound_obj=bound_obj,
+            message=message,
+        )
+        # Now return an awaitable that will finish the send.
+        return self._send_async_awaitable(
+            bound_obj, message, raw_response_awaitable
+        )
+
+    async def _send_async_awaitable(
+        self,
+        bound_obj: Any,
+        message: Message,
+        raw_response_awaitable: Awaitable[Response | SysResponse],
+    ) -> Response | None:
+        return self.unpack_raw_response(
+            bound_obj=bound_obj,
+            message=message,
+            raw_response=await raw_response_awaitable,
+        )
+
+    def fetch_raw_response(
+        self, bound_obj: Any, message: Message
+    ) -> Response | SysResponse:
+        """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.
+        """
+        if self._send_raw_message_call is None:
+            raise RuntimeError('send() is unimplemented for this type.')
+
+        msg_encoded = self._encode_message(bound_obj, message)
+        try:
+            response_encoded = self._send_raw_message_call(
+                bound_obj, msg_encoded
+            )
+        except Exception as exc:
+            response = ErrorSysResponse(
+                error_message='Error in MessageSender @send_method.',
+                error_type=(
+                    ErrorSysResponse.ErrorType.COMMUNICATION
+                    if isinstance(exc, CommunicationError)
+                    else ErrorSysResponse.ErrorType.LOCAL
+                ),
+            )
+            # Can include the actual exception since we'll be looking at
+            # this locally; might be helpful.
+            response.set_local_exception(exc)
+            return response
+        return self._decode_raw_response(bound_obj, message, response_encoded)
+
+    def fetch_raw_response_async(
+        self, bound_obj: Any, message: Message
+    ) -> Awaitable[Response | SysResponse]:
+        """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.
+        """
+
+        # Note: This call is synchronous so that the first part of it can
+        # happen synchronously. If the whole call were async we wouldn't be
+        # able to guarantee that messages sent in order would actually go
+        # out in order.
+        if (
+            self._send_async_raw_message_call is None
+            and self._send_async_raw_message_ex_call is None
+        ):
+            raise RuntimeError('send_async() is unimplemented for this type.')
+
+        msg_encoded = self._encode_message(bound_obj, message)
+        try:
+            if self._send_async_raw_message_ex_call is not None:
+                send_awaitable = self._send_async_raw_message_ex_call(
+                    bound_obj, msg_encoded, message
+                )
+            else:
+                assert self._send_async_raw_message_call is not None
+                send_awaitable = self._send_async_raw_message_call(
+                    bound_obj, msg_encoded
+                )
+        except Exception as exc:
+            return self._error_awaitable(exc)
+
+        # Now return an awaitable to finish the job.
+        return self._fetch_raw_response_awaitable(
+            bound_obj, message, send_awaitable
+        )
+
+    async def _error_awaitable(self, exc: Exception) -> SysResponse:
+        response = ErrorSysResponse(
+            error_message='Error in MessageSender @send_async_method.',
+            error_type=(
+                ErrorSysResponse.ErrorType.COMMUNICATION
+                if isinstance(exc, CommunicationError)
+                else ErrorSysResponse.ErrorType.LOCAL
+            ),
+        )
+        # Can include the actual exception since we'll be looking at
+        # this locally; might be helpful.
+        response.set_local_exception(exc)
+        return response
+
+    async def _fetch_raw_response_awaitable(
+        self, bound_obj: Any, message: Message, send_awaitable: Awaitable[str]
+    ) -> Response | SysResponse:
+
+        try:
+            response_encoded = await send_awaitable
+        except Exception as exc:
+            response = ErrorSysResponse(
+                error_message='Error in MessageSender @send_async_method.',
+                error_type=(
+                    ErrorSysResponse.ErrorType.COMMUNICATION
+                    if isinstance(exc, CommunicationError)
+                    else ErrorSysResponse.ErrorType.LOCAL
+                ),
+            )
+            # Can include the actual exception since we'll be looking at
+            # this locally; might be helpful.
+            response.set_local_exception(exc)
+            return response
+        return self._decode_raw_response(bound_obj, message, response_encoded)
+
+    def unpack_raw_response(
+        self,
+        bound_obj: Any,
+        message: Message,
+        raw_response: Response | SysResponse,
+    ) -> Response | None:
+        """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.
+        """
+        response = self._unpack_raw_response(bound_obj, raw_response)
+        assert (
+            response is None
+            or type(response) in type(message).get_response_types()
+        )
+        return response
+
+    def _encode_message(self, bound_obj: Any, message: Message) -> str:
+        """Encode a message for sending."""
+        msg_dict = self.protocol.message_to_dict(message)
+        if self._encode_filter_call is not None:
+            self._encode_filter_call(bound_obj, message, msg_dict)
+        return self.protocol.encode_dict(msg_dict)
+
+    def _decode_raw_response(
+        self, bound_obj: Any, message: Message, response_encoded: str
+    ) -> Response | SysResponse:
+        """Create a Response from returned data.
+
+        These Responses may encapsulate things like remote errors and
+        should not be handed directly to users. _unpack_raw_response()
+        should be used to translate to special values like None or raise
+        Exceptions. This function itself should never raise Exceptions.
+        """
+        response: Response | SysResponse
+        try:
+            response_dict = self.protocol.decode_dict(response_encoded)
+            response = self.protocol.response_from_dict(response_dict)
+            if self._decode_filter_call is not None:
+                self._decode_filter_call(
+                    bound_obj, message, response_dict, response
+                )
+        except Exception as exc:
+            response = ErrorSysResponse(
+                error_message='Error decoding raw response.',
+                error_type=ErrorSysResponse.ErrorType.LOCAL,
+            )
+            # Since we'll be looking at this locally, we can include
+            # extra info for logging/etc.
+            response.set_local_exception(exc)
+        return response
+
+    def _unpack_raw_response(
+        self, bound_obj: Any, raw_response: Response | SysResponse
+    ) -> Response | None:
+        """Given a raw Response, unpacks to special values or Exceptions.
+
+        The result of this call is what should be passed to users.
+        For complex messaging situations such as response callbacks
+        operating across different threads, this last stage should be
+        run such that any raised Exception is active when the callback
+        fires; not on the thread where the message was sent.
+        """
+        # EmptySysResponse translates to None
+        if isinstance(raw_response, EmptySysResponse):
+            return None
+
+        # Some error occurred. Raise a local Exception for it.
+        if isinstance(raw_response, ErrorSysResponse):
+
+            # Errors that happened locally can attach their exceptions
+            # here for extra logging goodness.
+            local_exception = raw_response.get_local_exception()
+
+            if (
+                raw_response.error_type
+                is ErrorSysResponse.ErrorType.COMMUNICATION
+            ):
+                raise CommunicationError(
+                    raw_response.error_message
+                ) from local_exception
+
+            # If something went wrong on *our* end of the connection,
+            # don't say it was a remote error.
+            if raw_response.error_type is ErrorSysResponse.ErrorType.LOCAL:
+                raise RuntimeError(
+                    raw_response.error_message
+                ) from local_exception
+
+            # If they want to support clean errors, do those.
+            if (
+                self.protocol.forward_clean_errors
+                and raw_response.error_type
+                is ErrorSysResponse.ErrorType.REMOTE_CLEAN
+            ):
+                raise CleanError(
+                    raw_response.error_message
+                ) from local_exception
+
+            if (
+                self.protocol.forward_communication_errors
+                and raw_response.error_type
+                is ErrorSysResponse.ErrorType.REMOTE_COMMUNICATION
+            ):
+                raise CommunicationError(
+                    raw_response.error_message
+                ) from local_exception
+
+            # Everything else gets lumped in as a remote error.
+            raise RemoteError(
+                raw_response.error_message,
+                peer_desc=(
+                    'peer'
+                    if self._peer_desc_call is None
+                    else self._peer_desc_call(bound_obj)
+                ),
+            ) from local_exception
+
+        assert isinstance(raw_response, Response)
+        return raw_response
+
+
+class BoundMessageSender:
+    """Base class for bound senders."""
+
+    def __init__(self, obj: Any, sender: MessageSender) -> None:
+        # Note: not checking obj here since we want to support
+        # at least our protocol property when accessed via type.
+        self._obj = obj
+        self._sender = sender
+
+    @property
+    def protocol(self) -> MessageProtocol:
+        """Protocol associated with this sender."""
+        return self._sender.protocol
+
+    def send_untyped(self, message: Message) -> Response | None:
+        """Send a message synchronously.
+
+        Whenever possible, use the send() call provided by generated
+        subclasses instead of this; it will provide better type safety.
+        """
+        assert self._obj is not None
+        return self._sender.send(bound_obj=self._obj, message=message)
+
+    def send_async_untyped(
+        self, message: Message
+    ) -> Awaitable[Response | None]:
+        """Send a message asynchronously.
+
+        Whenever possible, use the send_async() call provided by generated
+        subclasses instead of this; it will provide better type safety.
+        """
+        assert self._obj is not None
+        return self._sender.send_async(bound_obj=self._obj, message=message)
+
+    def fetch_raw_response_async_untyped(
+        self, message: Message
+    ) -> Awaitable[Response | SysResponse]:
+        """Split send (part 1 of 2)."""
+        assert self._obj is not None
+        return self._sender.fetch_raw_response_async(
+            bound_obj=self._obj, message=message
+        )
+
+    def unpack_raw_response_untyped(
+        self, message: Message, raw_response: Response | SysResponse
+    ) -> Response | None:
+        """Split send (part 2 of 2)."""
+        return self._sender.unpack_raw_response(
+            bound_obj=self._obj, message=message, raw_response=raw_response
+        )