AsyncioEventLoop

Complexity

Total Complexity

41

Size/Duplication

Total Lines	152
Duplicated Lines	0 %

Test Coverage

Coverage

71.43%

Importance

Changes	4
Bugs	0	Features	0

21 Methods

Rating	Name	Duplication	Size	Complexity
A	_setup_signals()	0	9	3
A	pipe_connection_lost()	0	3	2
A	_init()	0	7	2
A	_connect_socket()	0	6	2
A	process_exited()	0	3	1
A	_stop()	0	2	1
A	_stdin_reader()	0	6	2
A	_teardown_signals()	0	3	2
A	pipe_data_received()	0	8	3
B	_connect_stdio()	0	30	4
A	_start_reading()	0	2	1
A	connection_lost()	0	3	2
A	_connect_child()	0	5	1
A	_send()	0	7	2
A	_close()	0	6	2
A	_threadsafe_call()	0	2	1
A	_on_stdin()	0	6	3
A	_connect_tcp()	0	3	1
A	_run()	0	4	2
A	connection_made()	0	6	2
A	data_received()	0	6	2

"""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 logging
import os
import sys
from collections import deque
import threading

try:
    # For python 3.4+, use the standard library module
    import asyncio
except (ImportError, SyntaxError):
    # Fallback to trollius
    import trollius as asyncio

The import trollius could not be resolved.

from .base import BaseEventLoop

logger = logging.getLogger(__name__)
debug, info, warn = (logger.debug, logger.info, logger.warning,)

loop_cls = asyncio.SelectorEventLoop
if os.name == 'nt':
    # On windows use ProactorEventLoop which support pipes and is backed by the
    # more powerful IOCP facility
    # NOTE: we override in the stdio case, because it doesn't work.
    loop_cls = asyncio.ProactorEventLoop

    import msvcrt

The import msvcrt could not be resolved.

    from ctypes import windll, byref, wintypes, GetLastError, WinError, POINTER

Unused GetLastError imported from ctypes

    from ctypes.wintypes import HANDLE, DWORD, BOOL

    LPDWORD = POINTER(DWORD)

    PIPE_NOWAIT = wintypes.DWORD(0x00000001)

    ERROR_NO_DATA = 232

    def pipe_no_wait(pipefd):
        SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState
        SetNamedPipeHandleState.argtypes = [HANDLE, LPDWORD, LPDWORD, LPDWORD]
        SetNamedPipeHandleState.restype = BOOL

        h = msvcrt.get_osfhandle(pipefd)

        res = windll.kernel32.SetNamedPipeHandleState(h, byref(PIPE_NOWAIT), None, None)
        if res == 0:
            print(WinError())
            return False
        return True

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

The attribute _transport was defined outside __init__.

        self._raw_transport = transport

The attribute _raw_transport was defined outside __init__.

        if isinstance(transport, asyncio.SubprocessTransport):
            self._transport = transport.get_pipe_transport(0)

The attribute _transport was defined outside __init__.

    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)

The Instance of AsyncioEventLoop does not seem to have a member named _on_stderr.

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

The attribute _loop was defined outside __init__.

        self._queued_data = deque()

The attribute _queued_data was defined outside __init__.

        self._fact = lambda: self

The attribute _fact was defined outside __init__.

        self._raw_transport = None

The attribute _raw_transport was defined outside __init__.

        self._raw_stdout = False

The attribute _raw_stdout was defined outside __init__.

        self._active = False

The attribute _active was defined outside __init__.

    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 _on_stdin(self):
        while True:
            data = sys.stdin.buffer.read(1)
            if len(data) == 0:
                break
            self.data_received(data)

    def _stdin_reader(self):
        debug("started reader thread")
        while True: #self._active
            data = self._stdin.read(1)
            debug("reader thread read %d", len(data))
            self._loop.call_soon_threadsafe(self.data_received, data)

    def _connect_stdio(self):
        try:
            coroutine = self._loop.connect_read_pipe(self._fact, sys.stdin)
            self._loop.run_until_complete(coroutine)
            debug("native stdin connection successful")
        except OSError:
            debug("native stdin connection failed, using reader thread")
            if os.name == "nt":
                pass
                #pipe_no_wait(sys.stdin.fileno())
            else:
                import fcntl

The variable fcntl seems to be unused.

                #orig_fl = fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL)
                #fcntl.fcntl(sys.stdin.fileno(), fcntl.F_SETFL, orig_fl | os.O_NONBLOCK)
            #coro = self._loop.run_in_executor(None, self._stdin_reader)
            #self._loop.create_task(coro)
            self._stdin = sys.stdin.buffer

The attribute _stdin was defined outside __init__.

            self._active = True

The attribute _active was defined outside __init__.

            threading.Thread(target=self._stdin_reader).start()


        try:
            coroutine = self._loop.connect_write_pipe(self._fact, sys.stdout)
            self._loop.run_until_complete(coroutine)
            debug("native stdout connection successful")
        except OSError:
            debug("native stdin connection failed, using reader thread")

            self._stdout = sys.stdout.buffer

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

The attribute _stdout was defined outside __init__.

            self._raw_stdout = True

The attribute _raw_stdout was defined outside __init__.

    def _connect_child(self, argv):
        self._child_watcher = asyncio.get_child_watcher()

The attribute _child_watcher was defined outside __init__.

        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):
        debug("sent %s %s", repr(data), str(self._raw_stdout))
        if self._raw_stdout:
            self._stdout.write(data)
            self._stdout.flush()
        else:
            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):
        if self._raw_transport is not None:
            self._raw_transport.close()
        # FIXME: this is racy and stuff
        self._active = False

The attribute _active was defined outside __init__.

        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 = []

The attribute _signals was defined outside __init__.

            return

        self._signals = list(signals)

The attribute _signals was defined outside __init__.

        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)