LoaferManager

Complexity

Total Complexity

7

Size/Duplication

Total Lines	60
Duplicated Lines	0 %

Importance

Changes	9
Bugs	0	Features	1

5 Methods

Rating	Name	Duplication	Size	Complexity
A	start()	0	13	1
A	__init__()	0	21	2
A	on_future__errors()	0	6	2
A	stop()	0	12	1
A	get_dispatcher()	0	2	1

# -*- coding: utf-8 -*-
# vi:si:et:sw=4:sts=4:ts=4

import asyncio
from concurrent.futures import ThreadPoolExecutor
import logging
import os
import signal

from . import __version__
from .dispatcher import LoaferDispatcher

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

from .aws.consumer import Consumer as AWSConsumer

The name consumer does not seem to exist in module loafer.aws.

Unable to import 'aws.consumer' (invalid syntax (<string>, line 29))

logger = logging.getLogger(__name__)


class LoaferManager(object):

    def __init__(self, source, thread_pool_size=None, event_loop=None, consumers=None):

        self._loop = event_loop or asyncio.get_event_loop()
        self._loop.add_signal_handler(signal.SIGINT, self.stop)
        self._loop.add_signal_handler(signal.SIGTERM, self.stop)

        # XXX: See https://github.com/python/asyncio/issues/258
        # The minimum value depends on the number of cores in the machine
        # See https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor
        self.thread_pool_size = thread_pool_size
        self._executor = ThreadPoolExecutor(self.thread_pool_size)
        self._loop.set_default_executor(self._executor)

        self.routes = []

        if consumers is None:
            self.consumers = [AWSConsumer(source, {'WaitTimeSeconds': 5, 'MaxNumberOfMessages': 5}, loop=self._loop)]
        else:
            self.consumers = consumers

        self._dispatcher = None

    def get_dispatcher(self):
        return LoaferDispatcher(self.routes, self.consumers)

    def start(self):
        start_message = 'Starting Loafer - Version: {} (pid={}) ...'
        logger.info(start_message.format(__version__, os.getpid()))

        self._dispatcher = self.get_dispatcher()

        self._future = asyncio.gather(self._dispatcher.dispatch_consumers())

The attribute _future was defined outside __init__.

        self._future.add_done_callback(self.on_future__errors)

        try:
            self._loop.run_forever()
        finally:
            self._loop.close()

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

The argument kwargs seems to be unused.

The argument args seems to be unused.

        logger.info('Stopping Loafer ...')

        logger.debug('Stopping consumers ...')
        self._dispatcher.stop_consumers()

        logger.debug('Cancel schedulled operations ...')
        self._future.cancel()

        logger.debug('Waiting to shutdown ...')
        self._executor.shutdown(wait=True)
        self._loop.stop()

    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.stop()