UvEventLoop._send() - 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

UvEventLoop._send() A

↳ Parent: UvEventLoop

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 pyuv(libuv-python bindings)."""
import sys
from collections import deque

import pyuv
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from .base import BaseEventLoop


class UvEventLoop(BaseEventLoop):

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

    def _init(self):
        self._loop = pyuv.Loop()
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._async = pyuv.Async(self._loop, self._on_async)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._connection_error = None
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._error_stream = None
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._callbacks = deque()
class Foo:
    def __init__(self, x=None):
        self.x = x

    def _on_connect(self, stream, error):
        self.stop()
        if error:
            msg = 'Cannot connect to {}: {}'.format(
                self._connect_address, pyuv.errno.strerror(error))
            self._connection_error = IOError(msg)
class Foo:
    def __init__(self, x=None):
        self.x = x
            return
        self._read_stream = self._write_stream = stream
class Foo:
    def __init__(self, x=None):
        self.x = x

    def _on_read(self, handle, data, error):
        if error or not data:
            msg = pyuv.errno.strerror(error) if error else 'EOF'
            self._on_error(msg)
            return
        if handle == self._error_stream:
            return
        self._on_data(data)

    def _on_write(self, handle, error):

        if error:
            msg = pyuv.errno.strerror(error)
            self._on_error(msg)

    def _on_exit(self, handle, exit_status, term_signal):

        self._on_error('EOF')

    def _disconnected(self, *args):
class Foo:
    def some_method(self, x, y):
        return x + y;
        raise IOError('Not connected to Nvim')

    def _connect_tcp(self, address, port):
        stream = pyuv.TCP(self._loop)
        self._connect_address = '{}:{}'.format(address, port)
class Foo:
    def __init__(self, x=None):
        self.x = x
        stream.connect((address, port), self._on_connect)

    def _connect_socket(self, path):
        stream = pyuv.Pipe(self._loop)
        self._connect_address = path
class Foo:
    def __init__(self, x=None):
        self.x = x
        stream.connect(path, self._on_connect)

    def _connect_stdio(self):
        self._read_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._read_stream.open(sys.stdin.fileno())
        self._write_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._write_stream.open(sys.stdout.fileno())

    def _connect_child(self, argv):
        self._write_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._read_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._error_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        stdin = pyuv.StdIO(self._write_stream,
                           flags=pyuv.UV_CREATE_PIPE + pyuv.UV_READABLE_PIPE)
        stdout = pyuv.StdIO(self._read_stream,
                            flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
        stderr = pyuv.StdIO(self._error_stream,
                            flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
        pyuv.Process.spawn(self._loop,
                           args=argv,
                           exit_callback=self._on_exit,
                           flags=pyuv.UV_PROCESS_WINDOWS_HIDE,
                           stdio=(stdin, stdout, stderr,))
        self._error_stream.start_read(self._on_read)

    def _start_reading(self):
        if self._transport_type in ['tcp', 'socket']:
            self._loop.run()
            if self._connection_error:
                self.run = self.send = self._disconnected
                raise self._connection_error

        self._read_stream.start_read(self._on_read)

    def _send(self, data):
        self._write_stream.write(data, self._on_write)

    def _run(self):
        self._loop.run(pyuv.UV_RUN_DEFAULT)

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

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

    def _threadsafe_call(self, fn):
        self._callbacks.append(fn)
        self._async.send()

    def _on_async(self, handle):

        while self._callbacks:
            self._callbacks.popleft()()

    def _setup_signals(self, signals):
        self._signal_handles = []
class Foo:
    def __init__(self, x=None):
        self.x = x

        def handler(h, signum):

            self._on_signal(signum)

        for signum in signals:
            handle = pyuv.Signal(self._loop)
            handle.start(handler, signum)
            self._signal_handles.append(handle)

    def _teardown_signals(self):
        for handle in self._signal_handles:
            handle.stop()


1		"""Event loop implementation that uses pyuv(libuv-python bindings)."""
2	5	import sys
3	5	from collections import deque
4
5	5	import pyuv
		0 ignored issues – show Configuration introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The import `pyuv` 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...
6
7		from .base import BaseEventLoop
8
9
10		class UvEventLoop(BaseEventLoop):
11
12		"""`BaseEventLoop` subclass that uses `pvuv` as a backend."""
13
14		def _init(self):
15		self._loop = pyuv.Loop()
		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...
16		self._async = pyuv.Async(self._loop, self._on_async)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_async` 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...
17		self._connection_error = None
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connection_error` 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...
18		self._error_stream = None
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_error_stream` 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...
19		self._callbacks = deque()
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_callbacks` 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...
20
21		def _on_connect(self, stream, error):
22		self.stop()
23		if error:
24		msg = 'Cannot connect to {}: {}'.format(
25		self._connect_address, pyuv.errno.strerror(error))
26		self._connection_error = IOError(msg)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connection_error` 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...
27		return
28		self._read_stream = self._write_stream = stream
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_write_stream` 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... Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_read_stream` 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...
29
30		def _on_read(self, handle, data, error):
31		if error or not data:
32		msg = pyuv.errno.strerror(error) if error else 'EOF'
33		self._on_error(msg)
34		return
35		if handle == self._error_stream:
36		return
37		self._on_data(data)
38
39		def _on_write(self, handle, error):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `handle` seems to be unused. Loading history...
40		if error:
41		msg = pyuv.errno.strerror(error)
42		self._on_error(msg)
43
44		def _on_exit(self, handle, exit_status, term_signal):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `handle` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `exit_status` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `term_signal` seems to be unused. Loading history...
45		self._on_error('EOF')
46
47		def _disconnected(self, *args):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `args` seems to be unused. Loading history... Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
48		raise IOError('Not connected to Nvim')
49
50		def _connect_tcp(self, address, port):
51		stream = pyuv.TCP(self._loop)
52		self._connect_address = '{}:{}'.format(address, port)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connect_address` 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...
53		stream.connect((address, port), self._on_connect)
54
55		def _connect_socket(self, path):
56		stream = pyuv.Pipe(self._loop)
57		self._connect_address = path
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connect_address` 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...
58		stream.connect(path, self._on_connect)
59
60		def _connect_stdio(self):
61		self._read_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_read_stream` 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...
62		self._read_stream.open(sys.stdin.fileno())
63		self._write_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_write_stream` 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...
64		self._write_stream.open(sys.stdout.fileno())
65
66		def _connect_child(self, argv):
67		self._write_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_write_stream` 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...
68		self._read_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_read_stream` 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...
69		self._error_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_error_stream` 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...
70		stdin = pyuv.StdIO(self._write_stream,
71		flags=pyuv.UV_CREATE_PIPE + pyuv.UV_READABLE_PIPE)
72		stdout = pyuv.StdIO(self._read_stream,
73		flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
74		stderr = pyuv.StdIO(self._error_stream,
75		flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
76		pyuv.Process.spawn(self._loop,
77		args=argv,
78		exit_callback=self._on_exit,
79		flags=pyuv.UV_PROCESS_WINDOWS_HIDE,
80		stdio=(stdin, stdout, stderr,))
81		self._error_stream.start_read(self._on_read)
82
83		def _start_reading(self):
84		if self._transport_type in ['tcp', 'socket']:
85		self._loop.run()
86		if self._connection_error:
87		self.run = self.send = self._disconnected
88		raise self._connection_error
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report Raising NoneType while only classes or instances are allowed Loading history...
89		self._read_stream.start_read(self._on_read)
90
91		def _send(self, data):
92		self._write_stream.write(data, self._on_write)
93
94		def _run(self):
95		self._loop.run(pyuv.UV_RUN_DEFAULT)
96
97		def _stop(self):
98		self._loop.stop()
99
100		def _close(self):
101		self._loop.close()
102
103		def _threadsafe_call(self, fn):
104		self._callbacks.append(fn)
105		self._async.send()
106
107		def _on_async(self, handle):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `handle` seems to be unused. Loading history...
108		while self._callbacks:
109		self._callbacks.popleft()()
110
111		def _setup_signals(self, signals):
112		self._signal_handles = []
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_signal_handles` 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...
113
114		def handler(h, signum):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `h` seems to be unused. Loading history...
115		self._on_signal(signum)
116
117		for signum in signals:
118		handle = pyuv.Signal(self._loop)
119		handle.start(handler, signum)
120		self._signal_handles.append(handle)
121
122		def _teardown_signals(self):
123		for handle in self._signal_handles:
124		handle.stop()
125

neovim / python-client

Pull Request — master (#303)

UvEventLoop._send() 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