/usr/lib/python3/dist-packages/aiohttp/signals.py is in python3-aiohttp 0.20.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | import asyncio
from itertools import count
class BaseSignal(list):
@asyncio.coroutine
def _send(self, *args, **kwargs):
for receiver in self:
res = receiver(*args, **kwargs)
if asyncio.iscoroutine(res) or isinstance(res, asyncio.Future):
yield from res
def copy(self):
raise NotImplementedError("copy() is forbidden")
def sort(self):
raise NotImplementedError("sort() is forbidden")
class Signal(BaseSignal):
"""Coroutine-based signal implementation.
To connect a callback to a signal, use any list method.
Signals are fired using the :meth:`send` coroutine, which takes named
arguments.
"""
def __init__(self, app):
super().__init__()
self._app = app
klass = self.__class__
self._name = klass.__module__ + ':' + klass.__qualname__
self._pre = app.on_pre_signal
self._post = app.on_post_signal
@asyncio.coroutine
def send(self, *args, **kwargs):
"""
Sends data to all registered receivers.
"""
ordinal = None
debug = self._app._debug
if debug:
ordinal = self._pre.ordinal()
yield from self._pre.send(ordinal, self._name, *args, **kwargs)
yield from self._send(*args, **kwargs)
if debug:
yield from self._post.send(ordinal, self._name, *args, **kwargs)
class DebugSignal(BaseSignal):
@asyncio.coroutine
def send(self, ordinal, name, *args, **kwargs):
yield from self._send(ordinal, name, *args, **kwargs)
class PreSignal(DebugSignal):
def __init__(self):
super().__init__()
self._counter = count(1)
def ordinal(self):
return next(self._counter)
class PostSignal(DebugSignal):
pass
|