LoaferRunner.__init__() - Code Metrics - Inspection of "Rename loafer.manager -> loafer.managers" - georgeyk/loafer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( dc3db7...c94c8e )

by George

created 2017-01-07 21:50 UTC

LoaferRunner.init() A

↳ Parent: LoaferRunner

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	9
rs	9.6666

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

logger = logging.getLogger(__name__)


class LoaferRunner:

    def __init__(self, loop=None, max_workers=None, on_stop_callback=None):
        self._on_stop_callback = on_stop_callback
        self._loop = loop or asyncio.get_event_loop()

        # 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._executor = ThreadPoolExecutor(max_workers)
        self._loop.set_default_executor(self._executor)

    def start(self, future=None, run_forever=True):
        start = 'Starting Loafer - (pid={} / run_forever={}) ...'
        logger.info(start.format(os.getpid(), run_forever))

        self._loop.add_signal_handler(signal.SIGINT, self.stop)
        self._loop.add_signal_handler(signal.SIGTERM, self.stop)

        try:
            if run_forever:
                self._loop.run_forever()
            else:
                self._loop.run_until_complete(future)
        finally:
            self._loop.close()

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

        logger.info('Stopping Loafer ...')
        if callable(self._on_stop_callback):
            self._on_stop_callback()

        self._executor.shutdown(wait=True)
        if self._loop.is_running():
            self._loop.stop()


1			import asyncio
2			from concurrent.futures import ThreadPoolExecutor
3			import logging
4			import os
5			import signal
6
7			logger = logging.getLogger(__name__)
8
9
10			class LoaferRunner:
11
12			def __init__(self, loop=None, max_workers=None, on_stop_callback=None):
13			self._on_stop_callback = on_stop_callback
14			self._loop = loop or asyncio.get_event_loop()
15
16			# XXX: See https://github.com/python/asyncio/issues/258
17			# The minimum value depends on the number of cores in the machine
18			# See https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor
19			self._executor = ThreadPoolExecutor(max_workers)
20			self._loop.set_default_executor(self._executor)
21
22			def start(self, future=None, run_forever=True):
23			start = 'Starting Loafer - (pid={} / run_forever={}) ...'
24			logger.info(start.format(os.getpid(), run_forever))
25
26			self._loop.add_signal_handler(signal.SIGINT, self.stop)
27			self._loop.add_signal_handler(signal.SIGTERM, self.stop)
28
29			try:
30			if run_forever:
31			self._loop.run_forever()
32			else:
33			self._loop.run_until_complete(future)
34			finally:
35			self._loop.close()
36
37			def stop(self, args, *kwargs):
			0 ignored issues – show Unused Code introduced 2017-01-07 21:52 UTC by Report Bug Copy Issue Report The argument `args` seems to be unused. Loading history... Unused Code introduced 2017-01-07 21:52 UTC by Report Bug Copy Issue Report The argument `kwargs` seems to be unused. Loading history...
38			logger.info('Stopping Loafer ...')
39			if callable(self._on_stop_callback):
40			self._on_stop_callback()
41
42			self._executor.shutdown(wait=True)
43			if self._loop.is_running():
44			self._loop.stop()
45

georgeyk / loafer

Push — master ( dc3db7...c94c8e )

LoaferRunner.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

LoaferRunner.init() A