LoaferManager

Complexity

Total Complexity

7

Size/Duplication

Total Lines	34
Duplicated Lines	0 %

Importance

Changes	11
Bugs	0	Features	1

5 Methods

Rating	Name	Duplication	Size	Complexity
A	run()	0	4	1
A	__init__()	0	4	1
A	on_future__errors()	0	6	2
A	on_loop__stop()	0	7	2
A	dispatcher()	0	3	1

import asyncio
import logging

from cached_property import cached_property

from .dispatcher import LoaferDispatcher

Unable to import 'dispatcher' (invalid syntax (<string>, line 41))

from .runners import LoaferRunner

logger = logging.getLogger(__name__)


class LoaferManager:

    def __init__(self, routes, consumers, runner=None):
        self.runner = runner or LoaferRunner(on_stop_callback=self.on_loop__stop)
        self.routes = routes
        self.consumers = consumers

    @cached_property
    def dispatcher(self):
        return LoaferDispatcher(self.routes, self.consumers)

    def run(self, forever=True):
        self._future = asyncio.gather(self.dispatcher.dispatch_providers())

The attribute _future was defined outside __init__.

        self._future.add_done_callback(self.on_future__errors)
        self.runner.start(self._future, run_forever=forever)

    #
    # Callbacks
    #

    def on_future__errors(self, future):
        exc = future.exception()
        # Unhandled errors crashes the event loop execution
        if isinstance(exc, BaseException):
            logger.critical('Fatal error caught: {!r}'.format(exc))
            self.runner.stop()

    def on_loop__stop(self, *args, **kwargs):

The argument args seems to be unused.

The argument kwargs seems to be unused.

        logger.debug('Stopping consumers ...')
        self.dispatcher.stop_providers()

        if hasattr(self, '_future'):
            logger.debug('Cancel schedulled operations ...')
            self._future.cancel()