Source code for efro.message._module

# 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


# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
  Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum