1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
# -*- coding: utf-8 -*-
from typing import Any, Callable, Dict, Tuple
from twisted.internet.defer import Deferred, ensureDeferred
from twisted.python.failure import Failure
from pyee.base import EventEmitter, PyeeException
try:
from asyncio import iscoroutine
except ImportError:
iscoroutine = None
__all__ = ["TwistedEventEmitter"]
class TwistedEventEmitter(EventEmitter):
"""An event emitter class which can run twisted coroutines and handle
returned Deferreds, in addition to synchronous blocking functions. For
example::
@ee.on('event')
@inlineCallbacks
def async_handler(*args, **kwargs):
yield returns_a_deferred()
or::
@ee.on('event')
async def async_handler(*args, **kwargs):
await returns_a_deferred()
When async handlers fail, Failures are first emitted on the ``failure``
event. If there are no ``failure`` handlers, the Failure's associated
exception is then emitted on the ``error`` event. If there are no ``error``
handlers, the exception is raised. For consistency, when handlers raise
errors synchronously, they're captured, wrapped in a Failure and treated
as an async failure. This is unlike the behavior of EventEmitter,
which have no special error handling.
For twisted coroutine event handlers, calling emit is non-blocking.
In other words, you do not have to await any results from emit, and the
coroutine is scheduled in a fire-and-forget fashion.
Similar behavior occurs for "sync" functions which return Deferreds.
"""
def __init__(self):
super(TwistedEventEmitter, self).__init__()
def _emit_run(
self,
f: Callable,
args: Tuple[Any, ...],
kwargs: Dict[str, Any],
) -> None:
d = None
try:
result = f(*args, **kwargs)
except Exception:
self.emit("failure", Failure())
else:
if iscoroutine and iscoroutine(result):
d: Deferred[Any] = ensureDeferred(result)
elif isinstance(result, Deferred):
d = result
else:
return
def errback(failure: Failure) -> None:
if failure:
self.emit("failure", failure)
d.addErrback(errback)
def _emit_handle_potential_error(self, event: str, error: Any) -> None:
if event == "failure":
if isinstance(error, Failure):
try:
error.raiseException()
except Exception as exc:
self.emit("error", exc)
elif isinstance(error, Exception):
self.emit("error", error)
else:
self.emit("error", PyeeException(f"Unexpected failure object: {error}"))
else:
(super(TwistedEventEmitter, self))._emit_handle_potential_error(
event, error
)
|
