dupeguru/hscommon/notify.py

# Copyright 2015 Hardcoded Software (http://www.hardcoded.net)

# This software is licensed under the "GPLv3" License as described in the "LICENSE" file,
# which should be included with this package. The terms are also available at
# http://www.gnu.org/licenses/gpl-3.0.html

"""Very simple inter-object notification system.

This module is a brain-dead simple notification system involving a :class:`Broadcaster` and a
:class:`Listener`. A listener can only listen to one broadcaster. A broadcaster can have multiple
listeners. If the listener is connected, whenever the broadcaster calls :meth:`~Broadcaster.notify`,
the method with the same name as the broadcasted message is called on the listener.
"""

from collections import defaultdict
from typing import Callable, DefaultDict, List


class Broadcaster:
    """Broadcasts messages that are received by all listeners."""

    def __init__(self):
        self.listeners = set()

    def add_listener(self, listener: "Listener") -> None:
        self.listeners.add(listener)

    def notify(self, msg: str) -> None:
        """Notify all connected listeners of ``msg``.

        That means that each listeners will have their method with the same name as ``msg`` called.
        """
        for listener in self.listeners.copy():  # listeners can change during iteration
            if listener in self.listeners:  # disconnected during notification
                listener.dispatch(msg)

    def remove_listener(self, listener: "Listener") -> None:
        self.listeners.discard(listener)


class Listener:
    """A listener is initialized with the broadcaster it's going to listen to. Initially, it is not connected."""

    def __init__(self, broadcaster: Broadcaster) -> None:
        self.broadcaster = broadcaster
        self._bound_notifications: DefaultDict[str, List[Callable]] = defaultdict(list)

    def bind_messages(self, messages: str, func: Callable) -> None:
        """Binds multiple message to the same function.

        Often, we perform the same thing on multiple messages. Instead of having the same function
        repeated again and agin in our class, we can use this method to bind multiple messages to
        the same function.
        """
        for message in messages:
            self._bound_notifications[message].append(func)

    def connect(self) -> None:
        """Connects the listener to its broadcaster."""
        self.broadcaster.add_listener(self)

    def disconnect(self) -> None:
        """Disconnects the listener from its broadcaster."""
        self.broadcaster.remove_listener(self)

    def dispatch(self, msg: str) -> None:
        if msg in self._bound_notifications:
            for func in self._bound_notifications[msg]:
                func()
        if hasattr(self, msg):
            method = getattr(self, msg)
            method()


class Repeater(Broadcaster, Listener):
    REPEATED_NOTIFICATIONS = None

    def __init__(self, broadcaster: Broadcaster) -> None:
        Broadcaster.__init__(self)
        Listener.__init__(self, broadcaster)

    def _repeat_message(self, msg: str) -> None:
        if not self.REPEATED_NOTIFICATIONS or msg in self.REPEATED_NOTIFICATIONS:
            self.notify(msg)

    def dispatch(self, msg: str) -> None:
        Listener.dispatch(self, msg)
        self._repeat_message(msg)