PluginLoader

Complexity

Total Complexity

19

Size/Duplication

Total Lines	167
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	1

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__init__()	0	22	1
A	__repr__()	0	6	1
A	nreaders()	0	7	1
A	nwriters()	0	7	1
A	nplugins()	0	7	1
A	__str__()	0	6	1
B	load()	0	36	4
A	start()	0	13	3
B	stop()	0	26	5
A	__del__()	0	3	1

# -*- coding: utf-8 -*-
"""Plugin system oject that controls readers and writers.

.. moduleauthor:: Kenny Freeman <kenny.freeman@gmail.com>

"""
__author__ = 'Kenny Freeman'
__email__ = 'kenny.freeman@gmail.com'
__license__ = "ISCL"
__docformat__ = 'reStructuredText'

import os.path

import plumd
import plumd.config
import plumd.plugins
import plumd.plugins.util
import plumd.plugins.read
import plumd.plugins.write


class PluginLoader(object):
    """Loads and controls the reader and writer objects.

    raises:
        ConfigError if a plugin is configured with an invalid path
        ConfigError if a plugin configuration is missing 'name'
        DuplicatePlugin if a plugin configuration has a duplicate 'name'
        PluginLoadError if there was an error loading the plugin

    :param config: a simple configuration helper object.
    :type config: plumd.config.conf
    :param log: a logger object
    :type log: logger
    :raises: ConfigError, DuplicatePlugin, PluginLoadError
    """

    def __init__(self, log, config):
        """Loads and controls the reader and writer objects.

        raises:
            ConfigError if a plugin is configured with an invalid path
            ConfigError if a plugin configuration is missing 'name'
            DuplicatePlugin if a plugin configuration has a duplicate 'name'
            PluginLoadError if there was an error loading the plugin

        :param config: a simple configuration helper object.
        :type config: plumd.config.conf
        :param log: a logger object from structlog.get_logger()
        :type log: logger
        :raises: ConfigError, DuplicatePlugin, PluginLoadError
        """
        self.log = log
        msg = "initializing from: {0} conf: \n{1}"
        self.log.debug(msg.format(config.path, config))
        self.config = config
        self.readers = {} # reader plugin io objects
        self.writers = {} # writer plugin io objects
        self.load()


    def __str__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "readers={0}, writers={1}".format(list(self.readers.keys()),
                                              list(self.writers.keys()))
        return s


    def __repr__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "readers={0}, writers={1}".format(list(self.readers.keys()),
                                              list(self.writers.keys()))
        return s


    @property
    def nplugins(self):
        """Returns the number of loaded plugins.

        :rtype: int
        """
        return len(list(self.readers.keys())) + len(list(self.writers.keys()))


    @property
    def nreaders(self):
        """Return the number of loaded reader plugins.

        :rtype: int
        """
        return len(list(self.readers.keys()))


    @property
    def nwriters(self):
        """Return the number of loaded writer plugins.

        :rtype: int
        """
        return len(list(self.writers.keys()))


    def load(self):
        """Loads the configured plugin objects into self.readers and
        self.writers. Sets the writers for each reader plugin.

        raises:
            ConfigError if ptype has an incorrect configuration path set
            ConfigError if a plugin configuration is missing 'name'
            DuplicatePlugin if a plugin configuration has a duplicate 'name'
            PluginLoadError if there was an error loading a plugin

        :raises: ConfigError, DuplicatePlugin
        """
        # load all plugins
        pdir = self.config.get('config.plugins')
        if not os.path.isdir(pdir):
            msg = "invalid plugin directory configured: {0}"
            raise plumd.ConfigError(msg.format(pdir))
        msg = "loading plugins from: {0}"
        self.log.info(msg.format(self.config.get('config.plugins')))
        pobjs = plumd.plugins.util.load_all_plugins(self.log, self.config)
        # load_all_plugins returns a tuple of reader, writer objects
        robjs, wobjs = pobjs

        # setup reader threads for each reader plugin
        for rname, robj in robjs.items():
            clsn = plumd.plugins.read.PluginReader
            self.readers[rname] = clsn(self.log, robj, rname)

        # setup writer threads for each writer plugin
        for wname, wobj in wobjs.items():
            qsize = self.config.get('max.queue')
            clsn = plumd.plugins.write.PluginWriter
            self.writers[wname] = clsn(self.log, wobj, wname, qsize)

        # now the readers need to know what writers to write to
        plumd.plugins.util.config_plugin_writers(self)


    def start(self):
        """Calls the start() function on each reader/writer."""
        self.log.debug("starting")

        # start the writers then the readers
        plugins = [('writers', self.writers), ('readers', self.readers) ]
        for ptype, pdict in plugins:
            self.log.debug("starting: {0}".format(ptype))
            for pobj in pdict.values():
                pobj.start()

        # all done
        self.log.debug("started")


    def stop(self):
        """Calls the stop() function on each reader/writer."""
        self.log.debug("stopping plugins")

        # stop the readers first - they call each plugins onstop()
        self.log.debug("stopping readers")
        for pobj in list(self.readers.values()):
            self.log.debug("stopping reader: {0}".format(pobj.name))
            pobj.stop()
            pobj.onstop()

        ## next stop the writers - they call each plugins onstop()
        self.log.debug("stopping writers")
        for pobj in list(self.writers.values()):
            self.log.debug("stopping writer: {0}".format(pobj.name))
            pobj.stop()
            pobj.onstop()

        # join threads now
        for pdict in [self.readers, self.writers]:
            for pname, pobj in list(pdict.items()):
                self.log.debug("waiting for plugin to stop: {0}".format(pname))
                pobj.join(timeout=self.config.get('shutdown_timeout'))

        # all done
        self.log.debug("stopped")


    def __del__(self):
        """Log __del__ calls."""
        self.log.debug("del")