+++ /dev/null
-"""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
-
-try:
- # For python 3.4+, use the standard library module
- import asyncio
-except (ImportError, SyntaxError):
- # Fallback to trollius
- import trollius as asyncio
-
-from pynvim.msgpack_rpc.event_loop.base import BaseEventLoop
-
-logger = logging.getLogger(__name__)
-debug, info, warn = (logger.debug, logger.info, logger.warning,)
-
-loop_cls = asyncio.SelectorEventLoop
-if os.name == 'nt':
- from asyncio.windows_utils import PipeHandle
- import msvcrt
-
- # 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
-
-
-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
- self._raw_transport = transport
- if isinstance(transport, asyncio.SubprocessTransport):
- self._transport = transport.get_pipe_transport(0)
-
- 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()
- self._queued_data = deque()
- self._fact = lambda: self
- self._raw_transport = None
-
- 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):
- if os.name == 'nt':
- pipe = PipeHandle(msvcrt.get_osfhandle(sys.stdin.fileno()))
- else:
- pipe = sys.stdin
- coroutine = self._loop.connect_read_pipe(self._fact, pipe)
- self._loop.run_until_complete(coroutine)
- pass # replaces next logging statement
- #debug("native stdin connection successful")
-
- # Make sure subprocesses don't clobber stdout,
- # send the output to stderr instead.
- rename_stdout = os.dup(sys.stdout.fileno())
- os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
-
- if os.name == 'nt':
- pipe = PipeHandle(msvcrt.get_osfhandle(rename_stdout))
- else:
- pipe = os.fdopen(rename_stdout, 'wb')
- coroutine = self._loop.connect_write_pipe(self._fact, pipe)
- self._loop.run_until_complete(coroutine)
- pass # replaces next logging statement
- #debug("native stdout connection successful")
-
- def _connect_child(self, argv):
- if os.name != 'nt':
- self._child_watcher = asyncio.get_child_watcher()
- 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):
- if self._raw_transport is not None:
- self._raw_transport.close()
- 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 = []
- return
-
- self._signals = list(signals)
- 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)
projects / dotfiles / .git / blobdiff