+++ /dev/null
-"""Implements a Nvim host for python plugins."""
-import imp
-import inspect
-import logging
-import os
-import os.path
-import re
-from functools import partial
-from traceback import format_exc
-
-from pynvim.api import decode_if_bytes, walk
-from pynvim.compat import IS_PYTHON3, find_module
-from pynvim.msgpack_rpc import ErrorResponse
-from pynvim.plugin import script_host
-from pynvim.util import format_exc_skip, get_client_info
-
-__all__ = ('Host')
-
-logger = logging.getLogger(__name__)
-error, debug, info, warn = (logger.error, logger.debug, logger.info,
- logger.warning,)
-
-host_method_spec = {"poll": {}, "specs": {"nargs": 1}, "shutdown": {}}
-
-
-class Host(object):
-
- """Nvim host for python plugins.
-
- Takes care of loading/unloading plugins and routing msgpack-rpc
- requests/notifications to the appropriate handlers.
- """
-
- def __init__(self, nvim):
- """Set handlers for plugin_load/plugin_unload."""
- self.nvim = nvim
- self._specs = {}
- self._loaded = {}
- self._load_errors = {}
- self._notification_handlers = {
- 'nvim_error_event': self._on_error_event
- }
- self._request_handlers = {
- 'poll': lambda: 'ok',
- 'specs': self._on_specs_request,
- 'shutdown': self.shutdown
- }
-
- # Decode per default for Python3
- self._decode_default = IS_PYTHON3
-
- def _on_async_err(self, msg):
- # uncaught python exception
- self.nvim.err_write(msg, async_=True)
-
- def _on_error_event(self, kind, msg):
- # error from nvim due to async request
- # like nvim.command(..., async_=True)
- errmsg = "{}: Async request caused an error:\n{}\n".format(
- self.name, decode_if_bytes(msg))
- self.nvim.err_write(errmsg, async_=True)
- return errmsg
-
- def start(self, plugins):
- """Start listening for msgpack-rpc requests and notifications."""
- self.nvim.run_loop(self._on_request,
- self._on_notification,
- lambda: self._load(plugins),
- err_cb=self._on_async_err)
-
- def shutdown(self):
- """Shutdown the host."""
- self._unload()
- self.nvim.stop_loop()
-
- def _wrap_delayed_function(self, cls, delayed_handlers, name, sync,
- module_handlers, path, *args):
- # delete the delayed handlers to be sure
- for handler in delayed_handlers:
- method_name = handler._nvim_registered_name
- if handler._nvim_rpc_sync:
- del self._request_handlers[method_name]
- else:
- del self._notification_handlers[method_name]
- # create an instance of the plugin and pass the nvim object
- plugin = cls(self._configure_nvim_for(cls))
-
- # discover handlers in the plugin instance
- self._discover_functions(plugin, module_handlers, path, False)
-
- if sync:
- self._request_handlers[name](*args)
- else:
- self._notification_handlers[name](*args)
-
- def _wrap_function(self, fn, sync, decode, nvim_bind, name, *args):
- if decode:
- args = walk(decode_if_bytes, args, decode)
- if nvim_bind is not None:
- args.insert(0, nvim_bind)
- try:
- return fn(*args)
- except Exception:
- if sync:
- msg = ("error caught in request handler '{} {}':\n{}"
- .format(name, args, format_exc_skip(1)))
- raise ErrorResponse(msg)
- else:
- msg = ("error caught in async handler '{} {}'\n{}\n"
- .format(name, args, format_exc_skip(1)))
- self._on_async_err(msg + "\n")
-
- def _on_request(self, name, args):
- """Handle a msgpack-rpc request."""
- if IS_PYTHON3:
- name = decode_if_bytes(name)
- handler = self._request_handlers.get(name, None)
- if not handler:
- msg = self._missing_handler_error(name, 'request')
- pass # replaces next logging statement
- #error(msg)
- raise ErrorResponse(msg)
-
- pass # replaces next logging statement
- #debug('calling request handler for "%s", args: "%s"', name, args)
- rv = handler(*args)
- pass # replaces next logging statement
- #debug("request handler for '%s %s' returns: %s", name, args, rv)
- return rv
-
- def _on_notification(self, name, args):
- """Handle a msgpack-rpc notification."""
- if IS_PYTHON3:
- name = decode_if_bytes(name)
- handler = self._notification_handlers.get(name, None)
- if not handler:
- msg = self._missing_handler_error(name, 'notification')
- pass # replaces next logging statement
- #error(msg)
- self._on_async_err(msg + "\n")
- return
-
- pass # replaces next logging statement
- #debug('calling notification handler for "%s", args: "%s"', name, args)
- handler(*args)
-
- def _missing_handler_error(self, name, kind):
- msg = 'no {} handler registered for "{}"'.format(kind, name)
- pathmatch = re.match(r'(.+):[^:]+:[^:]+', name)
- if pathmatch:
- loader_error = self._load_errors.get(pathmatch.group(1))
- if loader_error is not None:
- msg = msg + "\n" + loader_error
- return msg
-
- def _load(self, plugins):
- has_script = False
- for path in plugins:
- err = None
- if path in self._loaded:
- pass # replaces next logging statement
- #error('{} is already loaded'.format(path))
- continue
- try:
- if path == "script_host.py":
- module = script_host
- has_script = True
- else:
- directory, name = os.path.split(os.path.splitext(path)[0])
- file, pathname, descr = find_module(name, [directory])
- module = imp.load_module(name, file, pathname, descr)
- handlers = []
- self._discover_classes(module, handlers, path)
- self._discover_functions(module, handlers, path, False)
- if not handlers:
- pass # replaces next logging statement
- #error('{} exports no handlers'.format(path))
- continue
- self._loaded[path] = {'handlers': handlers, 'module': module}
- except Exception as e:
- err = ('Encountered {} loading plugin at {}: {}\n{}'
- .format(type(e).__name__, path, e, format_exc(5)))
- pass # replaces next logging statement
- #error(err)
- self._load_errors[path] = err
-
- kind = ("script-host" if len(plugins) == 1 and has_script
- else "rplugin-host")
- info = get_client_info(kind, 'host', host_method_spec)
- self.name = info[0]
- self.nvim.api.set_client_info(*info, async_=True)
-
- def _unload(self):
- for path, plugin in self._loaded.items():
- handlers = plugin['handlers']
- for handler in handlers:
- method_name = handler._nvim_registered_name
- if hasattr(handler, '_nvim_shutdown_hook'):
- handler()
- elif handler._nvim_rpc_sync:
- del self._request_handlers[method_name]
- else:
- del self._notification_handlers[method_name]
- self._specs = {}
- self._loaded = {}
-
- def _discover_classes(self, module, handlers, plugin_path):
- for _, cls in inspect.getmembers(module, inspect.isclass):
- if getattr(cls, '_nvim_plugin', False):
- # discover handlers in the plugin instance
- self._discover_functions(cls, handlers, plugin_path, True)
-
- def _discover_functions(self, obj, handlers, plugin_path, delay):
- def predicate(o):
- return hasattr(o, '_nvim_rpc_method_name')
-
- cls_handlers = []
- specs = []
- objdecode = getattr(obj, '_nvim_decode', self._decode_default)
- for _, fn in inspect.getmembers(obj, predicate):
- method = fn._nvim_rpc_method_name
- if fn._nvim_prefix_plugin_path:
- method = '{}:{}'.format(plugin_path, method)
- sync = fn._nvim_rpc_sync
- if delay:
- fn_wrapped = partial(self._wrap_delayed_function, obj,
- cls_handlers, method, sync,
- handlers, plugin_path)
- else:
- decode = getattr(fn, '_nvim_decode', objdecode)
- nvim_bind = None
- if fn._nvim_bind:
- nvim_bind = self._configure_nvim_for(fn)
-
- fn_wrapped = partial(self._wrap_function, fn,
- sync, decode, nvim_bind, method)
- self._copy_attributes(fn, fn_wrapped)
- fn_wrapped._nvim_registered_name = method
- # register in the rpc handler dict
- if sync:
- if method in self._request_handlers:
- raise Exception(('Request handler for "{}" is '
- + 'already registered').format(method))
- self._request_handlers[method] = fn_wrapped
- else:
- if method in self._notification_handlers:
- raise Exception(('Notification handler for "{}" is '
- + 'already registered').format(method))
- self._notification_handlers[method] = fn_wrapped
- if hasattr(fn, '_nvim_rpc_spec'):
- specs.append(fn._nvim_rpc_spec)
- handlers.append(fn_wrapped)
- cls_handlers.append(fn_wrapped)
- if specs:
- self._specs[plugin_path] = specs
-
- def _copy_attributes(self, fn, fn2):
- # Copy _nvim_* attributes from the original function
- for attr in dir(fn):
- if attr.startswith('_nvim_'):
- setattr(fn2, attr, getattr(fn, attr))
-
- def _on_specs_request(self, path):
- if IS_PYTHON3:
- path = decode_if_bytes(path)
- if path in self._load_errors:
- self.nvim.out_write(self._load_errors[path] + '\n')
- return self._specs.get(path, 0)
-
- def _configure_nvim_for(self, obj):
- # Configure a nvim instance for obj (checks encoding configuration)
- nvim = self.nvim
- decode = getattr(obj, '_nvim_decode', self._decode_default)
- if decode:
- nvim = nvim.with_decode(decode)
- return nvim
projects / dotfiles / .git / blobdiff