AsyncioEventLoop._close() - Code Metrics - Inspection of "[WIP] closing the event loop" - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#303)

by Björn

created 2018-02-27 18:36 UTC

AsyncioEventLoop._close() A

↳ Parent: AsyncioEventLoop

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	2
ccs	0
cts	1
cp	0
crap	2
rs	10

"""Event loop implementation that uses the `asyncio` standard module.

The `asyncio` module was added to python standard library on 3.4, and it
provides a pure python implementation of an event loop library. It is used
as a fallback in case pyuv is not available(on python implementations other
than CPython).

Earlier python versions are supported through the `trollius` package, which
is a backport of `asyncio` that works on Python 2.6+.
"""
from __future__ import absolute_import

import os
import sys
from collections import deque

try:
    # For python 3.4+, use the standard library module
    import asyncio
except (ImportError, SyntaxError):
    # Fallback to trollius
    import trollius as asyncio
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from .base import BaseEventLoop

import warnings
warnings.simplefilter('default')

loop_cls = asyncio.SelectorEventLoop
if os.name == 'nt':
    # On windows use ProactorEventLoop which support pipes and is backed by the
    # more powerful IOCP facility
    loop_cls = asyncio.ProactorEventLoop


class AsyncioEventLoop(BaseEventLoop, asyncio.Protocol,
                       asyncio.SubprocessProtocol):

    """`BaseEventLoop` subclass that uses `asyncio` as a backend."""

    def connection_made(self, transport):
        """Used to signal `asyncio.Protocol` of a successful connection."""
        self._transport = transport
class Foo:
    def __init__(self, x=None):
        self.x = x
        if isinstance(transport, asyncio.SubprocessTransport):
            self._transport = transport.get_pipe_transport(0)
class Foo:
    def __init__(self, x=None):
        self.x = x

    def connection_lost(self, exc):
        """Used to signal `asyncio.Protocol` of a lost connection."""
        self._on_error(exc.args[0] if exc else 'EOF')

    def data_received(self, data):
        """Used to signal `asyncio.Protocol` of incoming data."""
        if self._on_data:
            self._on_data(data)
            return
        self._queued_data.append(data)

    def pipe_connection_lost(self, fd, exc):
        """Used to signal `asyncio.SubprocessProtocol` of a lost connection."""
        self._on_error(exc.args[0] if exc else 'EOF')

    def pipe_data_received(self, fd, data):
        """Used to signal `asyncio.SubprocessProtocol` of incoming data."""
        if fd == 2:  # stderr fd number
            self._on_stderr(data)

        elif self._on_data:
            self._on_data(data)
        else:
            self._queued_data.append(data)

    def process_exited(self):
        """Used to signal `asyncio.SubprocessProtocol` when the child exits."""
        self._on_error('EOF')

    def _init(self):
        self._loop = loop_cls()
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._loop.set_debug(True)
        self._queued_data = deque()
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._fact = lambda: self
class Foo:
    def __init__(self, x=None):
        self.x = x

    def _connect_tcp(self, address, port):
        coroutine = self._loop.create_connection(self._fact, address, port)
        self._loop.run_until_complete(coroutine)

    def _connect_socket(self, path):
        if os.name == 'nt':
            coroutine = self._loop.create_pipe_connection(self._fact, path)
        else:
            coroutine = self._loop.create_unix_connection(self._fact, path)
        self._loop.run_until_complete(coroutine)

    def _connect_stdio(self):
        coroutine = self._loop.connect_read_pipe(self._fact, sys.stdin)
        self._loop.run_until_complete(coroutine)
        coroutine = self._loop.connect_write_pipe(self._fact, sys.stdout)
        self._loop.run_until_complete(coroutine)

    def _connect_child(self, argv):
        self._child_watcher = asyncio.get_child_watcher()
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._child_watcher.attach_loop(self._loop)
        coroutine = self._loop.subprocess_exec(self._fact, *argv)
        self._loop.run_until_complete(coroutine)

    def _start_reading(self):
        pass

    def _send(self, data):
        self._transport.write(data)

    def _run(self):
        while self._queued_data:
            self._on_data(self._queued_data.popleft())
        self._loop.run_forever()

    def _stop(self):
        self._loop.stop()

    def _close(self):
        self._loop.close()

    def _threadsafe_call(self, fn):
        self._loop.call_soon_threadsafe(fn)

    def _setup_signals(self, signals):
        if os.name == 'nt':
            # add_signal_handler is not supported in win32
            self._signals = []
class Foo:
    def __init__(self, x=None):
        self.x = x
            return

        self._signals = list(signals)
class Foo:
    def __init__(self, x=None):
        self.x = x
        for signum in self._signals:
            self._loop.add_signal_handler(signum, self._on_signal, signum)

    def _teardown_signals(self):
        for signum in self._signals:
            self._loop.remove_signal_handler(signum)


1		"""Event loop implementation that uses the `asyncio` standard module.
2
3		The `asyncio` module was added to python standard library on 3.4, and it
4		provides a pure python implementation of an event loop library. It is used
5		as a fallback in case pyuv is not available(on python implementations other
6		than CPython).
7
8		Earlier python versions are supported through the `trollius` package, which
9		is a backport of `asyncio` that works on Python 2.6+.
10		"""
11	5	from __future__ import absolute_import
12
13	5	import os
14	5	import sys
15	5	from collections import deque
16
17	5	try:
18		# For python 3.4+, use the standard library module
19	5	import asyncio
20	3	except (ImportError, SyntaxError):
21		# Fallback to trollius
22	3	import trollius as asyncio
		0 ignored issues – show Configuration introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The import `trollius` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
23
24	5	from .base import BaseEventLoop
25
26		import warnings
27	5	warnings.simplefilter('default')
28	5
29		loop_cls = asyncio.SelectorEventLoop
30		if os.name == 'nt':
31		# On windows use ProactorEventLoop which support pipes and is backed by the
32		# more powerful IOCP facility
33		loop_cls = asyncio.ProactorEventLoop
34	5
35
36		class AsyncioEventLoop(BaseEventLoop, asyncio.Protocol,
37		asyncio.SubprocessProtocol):
38
39	5	"""`BaseEventLoop` subclass that uses `asyncio` as a backend."""
40
41	5	def connection_made(self, transport):
42	5	"""Used to signal `asyncio.Protocol` of a successful connection."""
43	5	self._transport = transport
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_transport` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
44		if isinstance(transport, asyncio.SubprocessTransport):
45	5	self._transport = transport.get_pipe_transport(0)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_transport` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
46
47		def connection_lost(self, exc):
48		"""Used to signal `asyncio.Protocol` of a lost connection."""
49	5	self._on_error(exc.args[0] if exc else 'EOF')
50
51		def data_received(self, data):
52		"""Used to signal `asyncio.Protocol` of incoming data."""
53		if self._on_data:
54		self._on_data(data)
55		return
56	5	self._queued_data.append(data)
57
58	4	def pipe_connection_lost(self, fd, exc):
59		"""Used to signal `asyncio.SubprocessProtocol` of a lost connection."""
60	5	self._on_error(exc.args[0] if exc else 'EOF')
61
62	5	def pipe_data_received(self, fd, data):
63	4	"""Used to signal `asyncio.SubprocessProtocol` of incoming data."""
64	5	if fd == 2: # stderr fd number
65	5	self._on_stderr(data)
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The Instance of `AsyncioEventLoop` does not seem to have a member named `_on_stderr`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
66		elif self._on_data:
67		self._on_data(data)
68		else:
69	5	self._queued_data.append(data)
70
71		def process_exited(self):
72		"""Used to signal `asyncio.SubprocessProtocol` when the child exits."""
73	5	self._on_error('EOF')
74	5
75	5	def _init(self):
76	5	self._loop = loop_cls()
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_loop` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
77		self._loop.set_debug(True)
78	5	self._queued_data = deque()
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_queued_data` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
79		self._fact = lambda: self
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_fact` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
80
81		def _connect_tcp(self, address, port):
82	5	coroutine = self._loop.create_connection(self._fact, address, port)
83		self._loop.run_until_complete(coroutine)
84
85		def _connect_socket(self, path):
86		if os.name == 'nt':
87		coroutine = self._loop.create_pipe_connection(self._fact, path)
88		else:
89	5	coroutine = self._loop.create_unix_connection(self._fact, path)
90		self._loop.run_until_complete(coroutine)
91
92		def _connect_stdio(self):
93	1	coroutine = self._loop.connect_read_pipe(self._fact, sys.stdin)
94		self._loop.run_until_complete(coroutine)
95	5	coroutine = self._loop.connect_write_pipe(self._fact, sys.stdout)
96	5	self._loop.run_until_complete(coroutine)
97	5
98		def _connect_child(self, argv):
99	5	self._child_watcher = asyncio.get_child_watcher()
		0 ignored issues – show Coding Style introduced 2017-01-08 20:41 UTC by Report Bug Copy Issue Report The attribute `_child_watcher` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
100	5	self._child_watcher.attach_loop(self._loop)
101		coroutine = self._loop.subprocess_exec(self._fact, *argv)
102	5	self._loop.run_until_complete(coroutine)
103	5
104		def _start_reading(self):
105	5	pass
106	5
107		def _send(self, data):
108	5	self._transport.write(data)
109
110	5	def _run(self):
111	5	while self._queued_data:
112		self._on_data(self._queued_data.popleft())
113	5	self._loop.run_forever()
114	5
115		def _stop(self):
116	5	self._loop.stop()
117	5
118		def _close(self):
119		self._loop.close()
120
121		def _threadsafe_call(self, fn):
122	5	self._loop.call_soon_threadsafe(fn)
123	5
124	5	def _setup_signals(self, signals):
125		if os.name == 'nt':
126	5	# add_signal_handler is not supported in win32
127	5	self._signals = []
		0 ignored issues – show Coding Style introduced 2016-02-15 16:00 UTC by Report Bug Copy Issue Report The attribute `_signals` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
128	5	return
129
130		self._signals = list(signals)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_signals` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
131		for signum in self._signals:
132		self._loop.add_signal_handler(signum, self._on_signal, signum)
133
134		def _teardown_signals(self):
135		for signum in self._signals:
136		self._loop.remove_signal_handler(signum)
137

neovim / python-client

Pull Request — master (#303)

AsyncioEventLoop._close() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files