• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# -*- coding: utf-8 -*-
2
3from functools import wraps
4from typing import Any, Callable, Dict, Optional, Tuple, Type, TypeVar, Union
5import warnings
6
7from typing_extensions import Literal
8
9from pyee.base import EventEmitter
10
11UpliftingEventEmitter = TypeVar(name="UpliftingEventEmitter", bound=EventEmitter)
12
13
14EMIT_WRAPPERS: Dict[EventEmitter, Callable[[], None]] = dict()
15
16
17def unwrap(event_emitter: EventEmitter) -> None:
18    """Unwrap an uplifted EventEmitter, returning it to its prior state."""
19    if event_emitter in EMIT_WRAPPERS:
20        EMIT_WRAPPERS[event_emitter]()
21
22
23def _wrap(
24    left: EventEmitter,
25    right: EventEmitter,
26    error_handler: Any,
27    proxy_new_listener: bool,
28) -> None:
29    left_emit = left.emit
30    left_unwrap: Optional[Callable[[], None]] = EMIT_WRAPPERS.get(left)
31
32    @wraps(left_emit)
33    def wrapped_emit(event: str, *args: Any, **kwargs: Any) -> bool:
34        left_handled: bool = left._call_handlers(event, args, kwargs)
35
36        # Do it for the right side
37        if proxy_new_listener or event != "new_listener":
38            right_handled = right._call_handlers(event, args, kwargs)
39        else:
40            right_handled = False
41
42        handled = left_handled or right_handled
43
44        # Use the error handling on ``error_handler`` (should either be
45        # ``left`` or ``right``)
46        if not handled:
47            error_handler._emit_handle_potential_error(event, args[0] if args else None)
48
49        return handled
50
51    def _unwrap() -> None:
52        warnings.warn(
53            DeprecationWarning(
54                "Patched ee.unwrap() is deprecated and will be removed in a "
55                "future release. Use pyee.uplift.unwrap instead."
56            )
57        )
58        unwrap(left)
59
60    def unwrap_hook() -> None:
61        left.emit = left_emit
62        if left_unwrap:
63            EMIT_WRAPPERS[left] = left_unwrap
64        else:
65            del EMIT_WRAPPERS[left]
66            del left.unwrap  # type: ignore
67        left.emit = left_emit
68
69        unwrap(right)
70
71    left.emit = wrapped_emit
72
73    EMIT_WRAPPERS[left] = unwrap_hook
74    left.unwrap = _unwrap  # type: ignore
75
76
77_PROXY_NEW_LISTENER_SETTINGS: Dict[str, Tuple[bool, bool]] = dict(
78    forward=(False, True),
79    backward=(True, False),
80    both=(True, True),
81    neither=(False, False),
82)
83
84
85ErrorStrategy = Union[Literal["new"], Literal["underlying"], Literal["neither"]]
86ProxyStrategy = Union[
87    Literal["forward"], Literal["backward"], Literal["both"], Literal["neither"]
88]
89
90
91def uplift(
92    cls: Type[UpliftingEventEmitter],
93    underlying: EventEmitter,
94    error_handling: ErrorStrategy = "new",
95    proxy_new_listener: ProxyStrategy = "forward",
96    *args: Any,
97    **kwargs: Any
98) -> UpliftingEventEmitter:
99    """A helper to create instances of an event emitter ``cls`` that inherits
100    event behavior from an ``underlying`` event emitter instance.
101
102    This is mostly helpful if you have a simple underlying event emitter
103    that you don't have direct control over, but you want to use that
104    event emitter in a new context - for example, you may want to ``uplift`` a
105    ``EventEmitter`` supplied by a third party library into an
106    ``AsyncIOEventEmitter`` so that you may register async event handlers
107    in your ``asyncio`` app but still be able to receive events from the
108    underlying event emitter and call the underlying event emitter's existing
109    handlers.
110
111    When called, ``uplift`` instantiates a new instance of ``cls``, passing
112    along any unrecognized arguments, and overwrites the ``emit`` method on
113    the ``underlying`` event emitter to also emit events on the new event
114    emitter and vice versa. In both cases, they return whether the ``emit``
115    method was handled by either emitter. Execution order prefers the event
116    emitter on which ``emit`` was called.
117
118    The ``unwrap`` function may be called on either instance; this will
119    unwrap both ``emit`` methods.
120
121    The ``error_handling`` flag can be configured to control what happens to
122    unhandled errors:
123
124    - 'new': Error handling for the new event emitter is always used and the
125      underlying library's non-event-based error handling is inert.
126    - 'underlying': Error handling on the underlying event emitter is always
127      used and the new event emitter can not implement non-event-based error
128      handling.
129    - 'neither': Error handling for the new event emitter is used if the
130      handler was registered on the new event emitter, and vice versa.
131
132    Tuning this option can be useful depending on how the underlying event
133    emitter does error handling. The default is 'new'.
134
135    The ``proxy_new_listener`` option can be configured to control how
136    ``new_listener`` events are treated:
137
138    - 'forward': ``new_listener`` events are propagated from the underlying
139    - 'both': ``new_listener`` events are propagated as with other events.
140    - 'neither': ``new_listener`` events are only fired on their respective
141      event emitters.
142      event emitter to the new event emitter but not vice versa.
143    - 'backward': ``new_listener`` events are propagated from the new event
144      emitter to the underlying event emitter, but not vice versa.
145
146    Tuning this option can be useful depending on how the ``new_listener``
147    event is used by the underlying event emitter, if at all. The default is
148    'forward', since ``underlying`` may not know how to handle certain
149    handlers, such as asyncio coroutines.
150
151    Each event emitter tracks its own internal table of handlers.
152    ``remove_listener``, ``remove_all_listeners`` and ``listeners`` all
153    work independently. This means you will have to remember which event
154    emitter an event handler was added to!
155
156    Note that both the new event emitter returned by ``cls`` and the
157    underlying event emitter should inherit from ``EventEmitter``, or at
158    least implement the interface for the undocumented ``_call_handlers`` and
159    ``_emit_handle_potential_error`` methods.
160    """
161
162    (
163        new_proxy_new_listener,
164        underlying_proxy_new_listener,
165    ) = _PROXY_NEW_LISTENER_SETTINGS[proxy_new_listener]
166
167    new: UpliftingEventEmitter = cls(*args, **kwargs)
168
169    uplift_error_handlers: Dict[str, Tuple[EventEmitter, EventEmitter]] = dict(
170        new=(new, new), underlying=(underlying, underlying), neither=(new, underlying)
171    )
172
173    new_error_handler, underlying_error_handler = uplift_error_handlers[error_handling]
174
175    _wrap(new, underlying, new_error_handler, new_proxy_new_listener)
176    _wrap(underlying, new, underlying_error_handler, underlying_proxy_new_listener)
177
178    return new
179