AsyncioEventLoop

Complexity

Total Complexity

36

Size/Duplication

Total Lines	107
Duplicated Lines	0 %

Test Coverage

Coverage

61.33%

Importance

Changes	2
Bugs	0	Features	0

20 Methods

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

"""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

The import trollius could not be resolved.

from .base import BaseEventLoop


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

The attribute _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__.

    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):
        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 _threadsafe_call(self, fn):
        self._loop.call_soon_threadsafe(fn)

    def _poll_fd(self, fd, on_readable, on_writable):
        if on_readable is not None:
            self._loop.add_reader(fd, on_readable)
        if on_writable is not None:
            self._loop.add_writer(fd, on_writable)
        def cancel():
            if on_readable is not None:
                self._loop.remove_reader(fd)
            if on_writable is not None:
                self._loop.remove_writer(fd)
        return cancel

    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)