Class EventDispatcher

redis/event.py:84–183

class EventDispatcher(EventDispatcherInterface):
    # TODO: Make dispatcher to accept external mappings.
    def __init__(
        self,
        event_listeners: Optional[
            Dict[Type[object], List[EventListenerInterface]]
        ] = None,
    ):
        """
        Dispatcher that dispatches events to listeners associated with given event.
        """
        self._event_listeners_mapping: Dict[
            Type[object], List[EventListenerInterface]
        ] = {
            AfterConnectionReleasedEvent: [
                ReAuthConnectionListener(),
            ],
            AfterPooledConnectionsInstantiationEvent: [
                RegisterReAuthForPooledConnections(),
            ],
            AfterSingleConnectionInstantiationEvent: [
                RegisterReAuthForSingleConnection()
            ],
            AfterPubSubConnectionInstantiationEvent: [RegisterReAuthForPubSub()],
            AfterAsyncClusterInstantiationEvent: [RegisterReAuthForAsyncClusterNodes()],
            AsyncAfterConnectionReleasedEvent: [
                AsyncReAuthConnectionListener(),
            ],
        }

        # Reentrant so a finalizer/listener that runs on the same thread
        # while the lock is held (e.g. a weakref.finalize callback fired
        # from cyclic GC during an allocation inside register_listeners /
        # unregister_listeners) can re-enter without deadlocking.
        self._lock = threading.RLock()
        self._async_lock = None

        if event_listeners:
            self.register_listeners(event_listeners)

    def dispatch(self, event: object):
        # Snapshot listeners under the lock, then release it before invoking
        # them. Holding the lock across listener execution would turn any
        # listener that calls register_listeners / unregister_listeners /
        # dispatch back into the dispatcher into a deadlock.
        with self._lock:
            listeners = list(self._event_listeners_mapping.get(type(event), []))
        for listener in listeners:
            listener.listen(event)

    async def dispatch_async(self, event: object):
        if self._async_lock is None:
            self._async_lock = asyncio.Lock()

        # Snapshot listeners under the lock, then release it before awaiting
        # them. See the note in dispatch(); the same rationale applies here
        # for dispatch_async re-entry from within a listener.
        async with self._async_lock: