extension_registry.py

# (C) Copyright 2007-2020 Enthought, Inc., Austin, TX
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in LICENSE.txt and may be redistributed only under
# the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
#
# Thanks for using Enthought open source!
""" A base class for extension registry implementation. """


# Standard library imports.
import functools
import logging
import types
import weakref

# Enthought library imports.
from traits.api import (
    ComparisonMode, Dict, HasTraits, provides, Str, Tuple,
)
from traits.observation.api import trait, observe

# Local imports.
from .extension_point_changed_event import ExtensionPointChangedEvent
from .i_extension_registry import IExtensionRegistry
from .unknown_extension_point import UnknownExtensionPoint


# Logging.
logger = logging.getLogger(__name__)


def _saferef(listener):
    """
    Weak reference for a (possibly bound method) listener.

    Returns a weakref.WeakMethod reference for bound methods,
    and a regular weakref.ref otherwise.

    This means that for example ``_saferef(myobj.mymethod)``
    returns a reference whose lifetime is connected to the
    lifetime of the object ``myobj``, rather than the lifetime
    of the temporary method ``myobj.mymethod``.

    Parameters
    ----------
    listener : callable
        Listener to return a weak reference for. This can be
        either a plain function, a bound method, or some other
        form of callable.

    Returns
    -------
    weakref.ref
        A weak reference to the listener. This will be a ``weakref.WeakMethod``
        object if the listener is an instance of ``types.MethodType``, and a
        plain ``weakref.ref`` otherwise.

    """
    if isinstance(listener, types.MethodType):
        return weakref.WeakMethod(listener)
    else:
        return weakref.ref(listener)


@provides(IExtensionRegistry)
class ExtensionRegistry(HasTraits):
    """ A base class for extension registry implementation. """

    ###########################################################################
    # Protected 'ExtensionRegistry' interface.
    ###########################################################################

    # A dictionary of extensions, keyed by extension point.
    # Mapping from extension point id (str) to a list of list of extensions
    # contributed to it.
    # Each item in the outer list is a list of extensions contributed by
    # a given plugin.
    _extensions = Dict()

    # The extension points that have been added *explicitly*.
    # Mapping from ExtensionPoint id (str) to ExtensionPoint
    _extension_points = Dict

    # Extension listeners.
    #
    # These are called when extensions are added to or removed from an
    # extension point.
    #
    # e.g. Dict(extension_point, [weakref.ref(callable)])
    #
    # A listener is any Python callable with the following signature:-
    #
    # def listener(extension_registry, extension_point_changed_event):
    #     ...
    _listeners = Dict

    ###########################################################################
    # 'IExtensionRegistry' interface.
    ###########################################################################

    def add_extension_point_listener(self, listener, extension_point_id=None):
        """ Add a listener for extensions being added or removed. """

        listeners = self._listeners.setdefault(extension_point_id, [])
        listeners.append(_saferef(listener))

        return

    def add_extension_point(self, extension_point):
        """ Add an extension point. """

        self._extension_points[extension_point.id] = extension_point
        logger.debug("extension point <%s> added", extension_point.id)

        return

    def get_extensions(self, extension_point_id):
        """ Return the extensions contributed to an extension point. """

        return self._get_extensions(extension_point_id)[:]

    def get_extension_point(self, extension_point_id):
        """ Return the extension point with the specified Id. """

        return self._extension_points.get(extension_point_id)

    def get_extension_points(self):
        """ Return all extension points. """

        return list(self._extension_points.values())

    def remove_extension_point_listener(
        self, listener, extension_point_id=None
    ):
        """ Remove a listener for extensions being added or removed. """

        listeners = self._listeners.setdefault(extension_point_id, [])
        listeners.remove(_saferef(listener))

        return

    def remove_extension_point(self, extension_point_id):
        """ Remove an extension point. """

        self._check_extension_point(extension_point_id)

        # Remove the extension point.
        del self._extension_points[extension_point_id]

        # Remove any extensions to the extension point.
        if extension_point_id in self._extensions:
            old = self._extensions[extension_point_id]
            del self._extensions[extension_point_id]

        else:
            old = []

        refs = self._get_listener_refs(extension_point_id)
        self._call_listeners(refs, extension_point_id, [], old, 0)

        logger.debug("extension point <%s> removed", extension_point_id)

        return

    def set_extensions(self, extension_point_id, extensions):
        """ Set the extensions contributed to an extension point. """

        self._check_extension_point(extension_point_id)

        old = self._get_extensions(extension_point_id)
        self._extensions[extension_point_id] = extensions

        refs = self._get_listener_refs(extension_point_id)
        self._call_listeners(refs, extension_point_id, extensions, old, None)

        return

    ###########################################################################
    # Protected 'ExtensionRegistry' interface.
    ###########################################################################

    def _call_listeners(self, refs, extension_point_id, added, removed, index):
        """ Call listeners that are listening to an extension point. """

        event = ExtensionPointChangedEvent(
            extension_point_id=extension_point_id,
            added=added,
            removed=removed,
            index=index,
        )

        for ref in refs:
            listener = ref()
            if listener is not None:
                listener(self, event)

        return

    def _check_extension_point(self, extension_point_id):
        """ Check to see if the extension point exists.

        Raise an 'UnknownExtensionPoint' if it does not.

        """

        if extension_point_id not in self._extension_points:
            raise UnknownExtensionPoint(extension_point_id)

        return

    def _get_extensions(self, extension_point_id):
        """ Return the extensions for the given extension point. """

        return self._extensions.setdefault(extension_point_id, [])

    def _get_listener_refs(self, extension_point_id):
        """ Get weak references to all listeners to an extension point.

        Returns a list containing the weak references to those listeners that
        are listening to this extension point specifically first, followed by
        those that are listening to any extension point.

        """

        refs = []
        refs.extend(self._listeners.get(extension_point_id, []))
        refs.extend(self._listeners.get(None, []))

        return refs


@provides(IExtensionRegistry)
class ObservableExtensionRegistry(HasTraits):
    """ A replacement for ExtensionRegistry to support Traits observe
    framework. Since ExtensionRegistry is a base class, this will have
    to be a base class too (?).

    Requires Traits 6.1+
    """

    # Mapping from extension point id (str) to a tuple(ExtensionPoint, List)
    _id_to_point_and_list = Dict(
        Str(),
        Tuple(),
        comparison_mode=ComparisonMode.identity,
    )

    # Mapping from extension point id to a callable to be used
    # for dispatching events from observe. Since the callable does not
    # hold a reference to the 'listener', it is okay for the dispatchers not
    # to be removed from this mapping after use, and it is okay for a
    # dispatcher to be reused many times for the same extension point id.
    _id_to_dispatcher = Dict()

    ###########################################################################
    # 'IExtensionRegistry' interface.
    ###########################################################################

    def add_extension_point_listener(self, listener, extension_point_id=None):
        """ Reimplemented IExtensionRegistry.add_extension_point_listener """

        observe(
            object=self,
            expression=(
                trait("_id_to_point_and_list", notify=False).dict_items()
            ),
            handler=listener,
            dispatcher=self._get_dispatcher(extension_point_id)
        )

    def add_extension_point(self, extension_point):
        """ Reimplemented IExtensionRegistry.add_extension_point """
        self._id_to_point_and_list[extension_point.id] = (
            extension_point, []
        )

    def get_extensions(self, extension_point_id):
        """ Reimplemented IExtensionRegistry.get_extensions """
        _, extensions = self._id_to_point_and_list.get(
            extension_point_id, (None, [])
        )
        return extensions[:]

    def get_extension_point(self, extension_point_id):
        """ Reimplemented IExtensionRegistry.get_extension_point """
        point, _ = self._id_to_point_and_list.get(
            extension_point_id, (None, [])
        )
        return point

    def get_extension_points(self):
        """ Reimplemented IExtensionRegistry.get_extension_points """
        return [point for point, _ in self._id_to_point_and_list.values()]

    def remove_extension_point_listener(
        self, listener, extension_point_id=None
    ):
        """ Reimplemented IExtensionRegistry.remove_extension_point_listener
        """
        expression = trait("_id_to_point_and_list", notify=False).dict_items()
        dispatcher = self._get_dispatcher(extension_point_id)
        try:
            observe(
                object=self,
                expression=expression,
                handler=listener,
                dispatcher=dispatcher,
                remove=True,
            )
        except Exception as exception:
            # Reraise ValueError to comply with interface requirement.
            raise ValueError(
                f"Unable to remove listener ({listener!r}) for extension "
                f"point id ({extension_point_id!r})"
            ) from exception

    def remove_extension_point(self, extension_point_id):
        """ Reimplemented IExtensionRegistry.remove_extension_point """
        try:
            del self._id_to_point_and_list[extension_point_id]
        except KeyError:
            raise UnknownExtensionPoint(extension_point_id)

    def set_extensions(self, extension_point_id, extensions):
        """ Reimplemented IExtensionRegistry.set_extensions """
        try:
            extension_point, _ = self._id_to_point_and_list[extension_point_id]
        except KeyError:
            raise UnknownExtensionPoint(extension_point_id)

        self._id_to_point_and_list[extension_point.id] = (
            extension_point, extensions
        )

    ###########################################################################
    # Private 'ObservableExtensionRegistry' interface.
    ###########################################################################

    def _get_dispatcher(self, extension_point_id):
        """ Return the callable for invoking the listener for the given
        extension point id, to be used as the 'dispatcher' argument in observe.

        The dispatcher is cached so that we can use it for removing the
        observer. It is not necessary to clear the cache and the cache can
        be reused indefinitely many times.

        Parameters
        ----------
        extension_point_id : str or None
            Extension point ID.

        Returns
        -------
        dispatcher : callable(listener, DictChangeEvent)
            Callable for invoking the listener.
        """
        if extension_point_id not in self._id_to_dispatcher:
            self._id_to_dispatcher[extension_point_id] = functools.partial(
                self._dispatch_listener,
                extension_point_id=extension_point_id,
            )
        return self._id_to_dispatcher[extension_point_id]

    def _dispatch_listener(self, listener, event, extension_point_id):
        """ Invoke the listener with the observed DictChangeEvent and
        extension point ID.

        Parameters
        ----------
        listener : callable(IExtensionRegistry, ExtensionPointChangedEvent)
            Listener to invoke.
        event : DictChangeEvent
            Change event caused by mutation to the internal extension point
            mapping.
        extension_point_id : str or None
            Specific extension point ID to observe.
        """
        if (extension_point_id is not None
                and extension_point_id not in event.added
                and extension_point_id not in event.removed):
            return

        _, added = event.added.get(extension_point_id, (None, []))
        _, removed = event.removed.get(extension_point_id, (None, []))
        listener(
            self,
            ExtensionPointChangedEvent(
                extension_point_id=extension_point_id,
                added=added,
                removed=removed,
                index=None,
            )
        )