--- /dev/null
+"""Synchronous msgpack-rpc session layer."""
+import logging
+import threading
+from collections import deque
+from traceback import format_exc
+
+import greenlet
+
+from pynvim.compat import check_async
+
+logger = logging.getLogger(__name__)
+error, debug, info, warn = (logger.error, logger.debug, logger.info,
+ logger.warning,)
+
+
+class Session(object):
+
+ """Msgpack-rpc session layer that uses coroutines for a synchronous API.
+
+ This class provides the public msgpack-rpc API required by this library.
+ It uses the greenlet module to handle requests and notifications coming
+ from Nvim with a synchronous API.
+ """
+
+ def __init__(self, async_session):
+ """Wrap `async_session` on a synchronous msgpack-rpc interface."""
+ self._async_session = async_session
+ self._request_cb = self._notification_cb = None
+ self._pending_messages = deque()
+ self._is_running = False
+ self._setup_exception = None
+ self.loop = async_session.loop
+ self._loop_thread = None
+
+ def threadsafe_call(self, fn, *args, **kwargs):
+ """Wrapper around `AsyncSession.threadsafe_call`."""
+ def handler():
+ try:
+ fn(*args, **kwargs)
+ except Exception:
+ pass # replaces next logging statement
+ #warn("error caught while excecuting async callback\n%s\n",
+ #format_exc())
+
+ def greenlet_wrapper():
+ gr = greenlet.greenlet(handler)
+ gr.switch()
+
+ self._async_session.threadsafe_call(greenlet_wrapper)
+
+ def next_message(self):
+ """Block until a message(request or notification) is available.
+
+ If any messages were previously enqueued, return the first in queue.
+ If not, run the event loop until one is received.
+ """
+ if self._is_running:
+ raise Exception('Event loop already running')
+ if self._pending_messages:
+ return self._pending_messages.popleft()
+ self._async_session.run(self._enqueue_request_and_stop,
+ self._enqueue_notification_and_stop)
+ if self._pending_messages:
+ return self._pending_messages.popleft()
+
+ def request(self, method, *args, **kwargs):
+ """Send a msgpack-rpc request and block until as response is received.
+
+ If the event loop is running, this method must have been called by a
+ request or notification handler running on a greenlet. In that case,
+ send the quest and yield to the parent greenlet until a response is
+ available.
+
+ When the event loop is not running, it will perform a blocking request
+ like this:
+ - Send the request
+ - Run the loop until the response is available
+ - Put requests/notifications received while waiting into a queue
+
+ If the `async_` flag is present and True, a asynchronous notification
+ is sent instead. This will never block, and the return value or error
+ is ignored.
+ """
+ async_ = check_async(kwargs.pop('async_', None), kwargs, False)
+ if async_:
+ self._async_session.notify(method, args)
+ return
+
+ if kwargs:
+ raise ValueError("request got unsupported keyword argument(s): {}"
+ .format(', '.join(kwargs.keys())))
+
+ if self._is_running:
+ v = self._yielding_request(method, args)
+ else:
+ v = self._blocking_request(method, args)
+ if not v:
+ # EOF
+ raise OSError('EOF')
+ err, rv = v
+ if err:
+ pass # replaces next logging statement
+ #info("'Received error: %s", err)
+ raise self.error_wrapper(err)
+ return rv
+
+ def run(self, request_cb, notification_cb, setup_cb=None):
+ """Run the event loop to receive requests and notifications from Nvim.
+
+ Like `AsyncSession.run()`, but `request_cb` and `notification_cb` are
+ inside greenlets.
+ """
+ self._request_cb = request_cb
+ self._notification_cb = notification_cb
+ self._is_running = True
+ self._setup_exception = None
+ self._loop_thread = threading.current_thread()
+
+ def on_setup():
+ try:
+ setup_cb()
+ except Exception as e:
+ self._setup_exception = e
+ self.stop()
+
+ if setup_cb:
+ # Create a new greenlet to handle the setup function
+ gr = greenlet.greenlet(on_setup)
+ gr.switch()
+
+ if self._setup_exception:
+ pass # replaces next logging statement
+ #error('Setup error: {}'.format(self._setup_exception))
+ raise self._setup_exception
+
+ # Process all pending requests and notifications
+ while self._pending_messages:
+ msg = self._pending_messages.popleft()
+ getattr(self, '_on_{}'.format(msg[0]))(*msg[1:])
+ self._async_session.run(self._on_request, self._on_notification)
+ self._is_running = False
+ self._request_cb = None
+ self._notification_cb = None
+ self._loop_thread = None
+
+ if self._setup_exception:
+ raise self._setup_exception
+
+ def stop(self):
+ """Stop the event loop."""
+ self._async_session.stop()
+
+ def close(self):
+ """Close the event loop."""
+ self._async_session.close()
+
+ def _yielding_request(self, method, args):
+ gr = greenlet.getcurrent()
+ parent = gr.parent
+
+ def response_cb(err, rv):
+ pass # replaces next logging statement
+ #debug('response is available for greenlet %s, switching back', gr)
+ gr.switch(err, rv)
+
+ self._async_session.request(method, args, response_cb)
+ pass # replaces next logging statement
+ #debug('yielding from greenlet %s to wait for response', gr)
+ return parent.switch()
+
+ def _blocking_request(self, method, args):
+ result = []
+
+ def response_cb(err, rv):
+ result.extend([err, rv])
+ self.stop()
+
+ self._async_session.request(method, args, response_cb)
+ self._async_session.run(self._enqueue_request,
+ self._enqueue_notification)
+ return result
+
+ def _enqueue_request_and_stop(self, name, args, response):
+ self._enqueue_request(name, args, response)
+ self.stop()
+
+ def _enqueue_notification_and_stop(self, name, args):
+ self._enqueue_notification(name, args)
+ self.stop()
+
+ def _enqueue_request(self, name, args, response):
+ self._pending_messages.append(('request', name, args, response,))
+
+ def _enqueue_notification(self, name, args):
+ self._pending_messages.append(('notification', name, args,))
+
+ def _on_request(self, name, args, response):
+ def handler():
+ try:
+ rv = self._request_cb(name, args)
+ pass # replaces next logging statement
+ #debug('greenlet %s finished executing, '
+ #+ 'sending %s as response', gr, rv)
+ response.send(rv)
+ except ErrorResponse as err:
+ pass # replaces next logging statement
+ #warn("error response from request '%s %s': %s", name,
+ #args, format_exc())
+ response.send(err.args[0], error=True)
+ except Exception as err:
+ pass # replaces next logging statement
+ #warn("error caught while processing request '%s %s': %s", name,
+ #args, format_exc())
+ response.send(repr(err) + "\n" + format_exc(5), error=True)
+ pass # replaces next logging statement
+ #debug('greenlet %s is now dying...', gr)
+
+ # Create a new greenlet to handle the request
+ gr = greenlet.greenlet(handler)
+ pass # replaces next logging statement
+ #debug('received rpc request, greenlet %s will handle it', gr)
+ gr.switch()
+
+ def _on_notification(self, name, args):
+ def handler():
+ try:
+ self._notification_cb(name, args)
+ pass # replaces next logging statement
+ #debug('greenlet %s finished executing', gr)
+ except Exception:
+ pass # replaces next logging statement
+ #warn("error caught while processing notification '%s %s': %s",
+ #name, args, format_exc())
+
+ pass # replaces next logging statement
+ #debug('greenlet %s is now dying...', gr)
+
+ gr = greenlet.greenlet(handler)
+ pass # replaces next logging statement
+ #debug('received rpc notification, greenlet %s will handle it', gr)
+ gr.switch()
+
+
+class ErrorResponse(BaseException):
+
+ """Raise this in a request handler to respond with a given error message.
+
+ Unlike when other exceptions are caught, this gives full control off the
+ error response sent. When "ErrorResponse(msg)" is caught "msg" will be
+ sent verbatim as the error response.No traceback will be appended.
+ """
+
+ pass
projects / dotfiles / .git / blobdiff