UvEventLoop._poll_fd()

Complexity

Conditions

6

Size

Total Lines

15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	42

Importance

Changes

0

1 Method

Rating	Name	Duplication	Size	Complexity
A	UvEventLoop.callback()	0	5	3

"""Event loop implementation that uses pyuv(libuv-python bindings)."""
import sys
from collections import deque

import pyuv

The import pyuv could not be resolved.

from .base import BaseEventLoop


class UvEventLoop(BaseEventLoop):

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

    def _init(self):
        self._loop = pyuv.Loop()

The attribute _loop was defined outside __init__.

        self._async = pyuv.Async(self._loop, self._on_async)

The attribute _async was defined outside __init__.

        self._connection_error = None

The attribute _connection_error was defined outside __init__.

        self._error_stream = None

The attribute _error_stream was defined outside __init__.

        self._callbacks = deque()

The attribute _callbacks was defined outside __init__.

    def _on_connect(self, stream, error):
        self.stop()
        if error:
            msg = 'Cannot connect to {0}: {1}'.format(
                self._connect_address, pyuv.errno.strerror(error))
            self._connection_error = IOError(msg)

The attribute _connection_error was defined outside __init__.

            return
        self._read_stream = self._write_stream = stream

The attribute _write_stream was defined outside __init__.

The attribute _read_stream was defined outside __init__.

    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):

The argument handle seems to be unused.

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

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

The argument handle seems to be unused.

The argument exit_status seems to be unused.

The argument term_signal seems to be unused.

        self._on_error('EOF')

    def _disconnected(self, *args):

The argument args seems to be unused.

This method could be written as a function/class method.

        raise IOError('Not connected to Nvim')

    def _connect_tcp(self, address, port):
        stream = pyuv.TCP(self._loop)
        self._connect_address = '{0}:{1}'.format(address, port)

The attribute _connect_address was defined outside __init__.

        stream.connect((address, port), self._on_connect)

    def _connect_socket(self, path):
        stream = pyuv.Pipe(self._loop)
        self._connect_address = path

The attribute _connect_address was defined outside __init__.

        stream.connect(path, self._on_connect)

    def _connect_stdio(self):
        self._read_stream = pyuv.Pipe(self._loop)

The attribute _read_stream was defined outside __init__.

        self._read_stream.open(sys.stdin.fileno())
        self._write_stream = pyuv.Pipe(self._loop)

The attribute _write_stream was defined outside __init__.

        self._write_stream.open(sys.stdout.fileno())

The Instance of RedirectStream does not seem to have a member named fileno.

    def _connect_child(self, argv):
        self._write_stream = pyuv.Pipe(self._loop)

The attribute _write_stream was defined outside __init__.

        self._read_stream = pyuv.Pipe(self._loop)

The attribute _read_stream was defined outside __init__.

        self._error_stream = pyuv.Pipe(self._loop)

The attribute _error_stream was defined outside __init__.

        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)
        self._process = pyuv.Process(self._loop)

The attribute _process was defined outside __init__.

        self._process.spawn(file=argv[0],
                            exit_callback=self._on_exit,
                            args=argv[1:],
                            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

Raising NoneType while only classes or instances are allowed

        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 _threadsafe_call(self, fn):
        self._callbacks.append(fn)
        self._async.send()

    def _on_async(self, handle):

The argument handle seems to be unused.

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

    def _poll_fd(self, fd, on_readable, on_writable):
        poll = pyuv.Poll(self._loop, fd)
        events = 0
        if on_readable is not None:
            events |= pyuv.UV_READABLE
        if on_writable is not None:
            events |= pyuv.UV_WRITABLE
        def callback(poll_handle, evts, errorno):

The argument poll_handle seems to be unused.

The argument errorno seems to be unused.

            if evts & pyuv.UV_READABLE:
                on_readable()
            if evts & pyuv.UV_WRITABLE:
                on_writable()

        poll.start(events, callback)
        return poll.stop

    def _setup_signals(self, signals):
        self._signal_handles = []

The attribute _signal_handles was defined outside __init__.

        def handler(h, signum):

The argument h seems to be unused.

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