src.fairreckitlib.core.events.event_dispatcher

  1"""This module contains a class that implements dispatcher/listener behaviour.
  2
  3Classes:
  4
  5    EventDispatcher: can dispatch an event to the respectively subscribed listeners.
  6
  7This program has been developed by students from the bachelor Computer Science at
  8Utrecht University within the Software Project course.
  9© Copyright Utrecht University (Department of Information and Computing Sciences)
 10"""
 11
 12from typing import Any, Callable, Optional, Tuple
 13
 14from .event_args import EventArgs
 15
 16
 17class EventDispatcher:
 18    """Event Dispatcher provides tools to communicate events to listeners.
 19
 20    The dispatcher is a centralized object that keeps track of all registered listeners
 21    for a specified event ID. These listeners can be added and removed dynamically.
 22
 23    Public methods:
 24
 25    add_listener
 26    get_num_listeners
 27    get_num_listeners_total
 28    remove_listener
 29    dispatch
 30    """
 31
 32    def __init__(self):
 33        """Construct the EventDispatcher."""
 34        self.listeners = {}
 35
 36    def add_listener(
 37            self,
 38            event_id: str,
 39            event_listener: Any,
 40            func_on_event: Tuple[
 41                Callable[[Any, EventArgs], None],
 42                Optional[Callable[[Any, EventArgs], None]]
 43            ]) -> None:
 44        """Add a listener for the specified event ID.
 45
 46        The event_listener and func_on_event are joined as a tuple to describe the listener
 47        and is expected to be unique. The listener will be notified by the dispatcher
 48        every time that the event arguments with the specified ID is propagated.
 49        This function raises a KeyError when the listener tuple already exists.
 50
 51        Args:
 52            event_id: unique ID that classifies the event.
 53            event_listener: the listener of the event. This object is passed to
 54                the event callback function as the first argument when the event is dispatched.
 55            func_on_event: the callback functions that are called when the event
 56                is dispatched. The first argument is the event_listener, followed by event args.
 57                and any keyword args
 58        """
 59        if event_id not in self.listeners:
 60            self.listeners[event_id] = []
 61
 62        listener = (event_listener, func_on_event)
 63        if listener in self.listeners[event_id]:
 64            raise KeyError('listener is already registered for event', event_id)
 65
 66        self.listeners[event_id].append(listener)
 67
 68    def get_num_listeners(self, event_id: str) -> Optional[int]:
 69        """Get the amount of listeners for the specified event id.
 70
 71        Args:
 72            event_id: the event id to query the number of listeners of.
 73
 74        Returns:
 75            the number of listeners of the event or None when the event is not registered.
 76        """
 77        return len(self.listeners[event_id]) if event_id in self.listeners else None
 78
 79    def get_num_listeners_total(self) -> int:
 80        """Get the total amount of listeners for the dispatcher.
 81
 82        Returns:
 83            the number of listeners in total.
 84        """
 85        num_listeners = 0
 86
 87        for event_id, _ in self.listeners.items():
 88            num_listeners += self.get_num_listeners(event_id)
 89
 90        return num_listeners
 91
 92    def remove_listener(
 93            self,
 94            event_id: str,
 95            event_listener: Any,
 96            func_on_event: Tuple[
 97                Callable[[Any, EventArgs], None],
 98                Optional[Callable[[Any, EventArgs], None]]
 99            ]) -> None:
100        """Remove a listener for the specified event ID.
101
102        The event_listener and func_on_event are joined as a tuple to describe the listener
103        and is expected to be unique. The listener is also expected to be identical to the one
104        that was used in 'add_listener' and will no longer be notified by the dispatcher when the
105        event arguments with the specified ID is propagated.
106        This function raises a KeyError when either the event_id or listener tuple does not exist.
107
108        Args:
109            event_id: unique ID that classifies the event.
110            event_listener: the listener of the event. This object is passed to
111                the event callback function as the first argument when the event is dispatched.
112            func_on_event: the callback functions that are called when the event
113                is dispatched. The first argument is the event_listener, followed by event args
114                and any keyword args.
115        """
116        if event_id not in self.listeners:
117            raise KeyError('event is not registered:', event_id)
118
119        listener = (event_listener, func_on_event)
120        if listener not in self.listeners[event_id]:
121            raise KeyError('listener is not registered for event:', event_id)
122
123        self.listeners[event_id].remove(listener)
124
125    def dispatch(self, event_args: EventArgs, **kwargs) -> bool:
126        """Dispatch event arguments with the corresponding event ID.
127
128        The event arguments with the specified ID will be propagated to all registered listeners.
129        For each listener their respective callback function is called with the listener
130        as the first argument, followed by the specified event args and keyword args.
131
132        Args:
133            event_args: the event's arguments.
134
135        Keyword Args:
136            Any: varies depending on the event.
137
138        Returns:
139            whether the event was dispatched to any registered listeners.
140        """
141        if event_args.event_id not in self.listeners:
142            return False
143
144        for (event_listener, func_on_event) in self.listeners[event_args.event_id]:
145            internal_func, external_func = func_on_event
146            internal_func(event_listener, event_args, **kwargs)
147            if external_func is not None:
148                external_func(event_listener, event_args, **kwargs)
149
150        return True