WIP: add _main.py

autoload/denite/init.vim

@@ -74,6 +74,13 @@ function! s:initialize_variables() abort
   let g:denite#_previewed_buffers = {}
   let g:denite#_ret = {}
   let g:denite#_async_ret = {}
+  let g:denite#_serveraddr =
+        \ denite#util#has_yarp() ?
+        \ neovim_rpc#serveraddr() : v:servername
+  if g:denite#_serveraddr ==# ''
+    " Use NVIM_LISTEN_ADDRESS
+    let g:denite#_serveraddr = $NVIM_LISTEN_ADDRESS
+  endif
 endfunction
 
 function! denite#init#_user_options() abort

rplugin/python3/denite/async.py → rplugin/python3/denite/aprocess.py

@@ -1,5 +1,5 @@
 # ============================================================================
-# FILE: async.py
+# FILE: aprocess.py
 # AUTHOR: Shougo Matsushita <Shougo.Matsu at gmail.com>
 # License: MIT license
 # ============================================================================

rplugin/python3/denite/child.py

@@ -9,8 +9,10 @@
     import_rplugins, expand, split_input, abspath)
 
 import copy
+import msgpack
 import os
 import re
+import sys
 import time
 from os.path import normpath, normcase
 from collections import ChainMap
@@ -26,6 +28,38 @@ def __init__(self, vim):
         self._kinds = {}
         self._runtimepath = ''
         self._current_sources = []
+        self._unpacker = msgpack.Unpacker(
+            encoding='utf-8',
+            unicode_errors='surrogateescape')
+        self._packer = msgpack.Packer(
+            use_bin_type=True,
+            encoding='utf-8',
+            unicode_errors='surrogateescape')
+
+    def main_loop(self, stdout):
+        while True:
+            feed = sys.stdin.buffer.raw.read(102400)
+            if feed is None:
+                continue
+            if feed == b'':
+                # EOF
+                return
+
+            self._unpacker.feed(feed)
+
+            for child_in in self._unpacker:
+                name = child_in['name']
+                args = child_in['args']
+                queue_id = child_in['queue_id']
+
+                self.debug('hello')
+                ret = self.main(name, args, queue_id)
+                if ret:
+                    # self._write(stdout, ret)
+                    _ret = self._vim.vars['denite#_ret']
+                    _ret[queue_id] = ret
+                    self._vim.vars['denite#_ret'] = _ret
+                    self.debug(_ret)
 
     def main(self, name, args, queue_id):
         ret = None

rplugin/python3/denite/parent.py

@@ -4,7 +4,17 @@
 # License: MIT license
 # ============================================================================
 
+import asyncio
 import time
+import os
+import msgpack
+import subprocess
+from functools import partial
+from pathlib import Path
+from queue import Queue
+
+from denite.aprocess import Process
+from denite.util import error_tb, error
 
 
 class _Parent(object):
@@ -31,16 +41,16 @@ def init_syntax(self, context, is_multi):
         self._put('init_syntax', [context, is_multi])
 
     def filter_candidates(self, context):
-        return self._put('filter_candidates', [context])
+        return self._get('filter_candidates', [context])
 
     def do_action(self, context, action_name, targets):
-        return self._put('do_action', [context, action_name, targets])
+        return self._get('do_action', [context, action_name, targets])
 
     def get_action(self, context, action_name, targets):
-        return self._put('get_action', [context, action_name, targets])
+        return self._get('get_action', [context, action_name, targets])
 
     def get_action_names(self, context, targets):
-        return self._put('get_action_names', [context, targets])
+        return self._get('get_action_names', [context, targets])
 
 
 class SyncParent(_Parent):
@@ -61,3 +71,110 @@ def _put(self, name, args):
         self._vim.vars['denite#_ret'] = _ret
 
         return self._vim.vars['denite#_ret'][queue_id]
+
+    def _get(self, name, args):
+        return self._put(name, args)
+
+
+class ASyncParent(_Parent):
+    def _start_process(self):
+        self._stdin = None
+        self._queue_id = ''
+        self._queue_in = Queue()
+        self._queue_out = Queue()
+        self._packer = msgpack.Packer(
+            use_bin_type=True,
+            encoding='utf-8',
+            unicode_errors='surrogateescape')
+        self._unpacker = msgpack.Unpacker(
+            encoding='utf-8',
+            unicode_errors='surrogateescape')
+
+        startupinfo = None
+        if os.name == 'nt':
+            startupinfo = subprocess.STARTUPINFO()
+            startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
+
+        main = str(Path(__file__).parent.parent.parent.parent.joinpath(
+            'autoload', 'denite', '_main.py'))
+
+        self._hnd = self._vim.loop.create_task(
+            self._vim.loop.subprocess_exec(
+                partial(Process, self),
+                self._vim.vars.get('python3_host_prog', 'python3'),
+                main,
+                self._vim.vars['denite#_serveraddr'],
+                stderr=None,
+                startupinfo=startupinfo))
+
+    def _connect_stdin(self, stdin):
+        self._stdin = stdin
+        return self._unpacker
+
+    def filter_candidates(self, context):
+        if self._queue_id:
+            # Use previous id
+            queue_id = self._queue_id
+        else:
+            queue_id = self._put('filter_candidates', [context])
+            if not queue_id:
+                return (False, [])
+
+        get = self._get(queue_id, True)
+        if not get:
+            # Skip the next merge_results
+            self._queue_id = queue_id
+            return [True, '', [], []]
+        self._queue_id = ''
+        results = get[0]
+        return results if results else [False, '', [], []]
+
+    def _put(self, name, args):
+        if not self._hnd:
+            return None
+
+        queue_id = str(time.time())
+        msg = self._packer.pack({
+            'name': name, 'args': args, 'queue_id': queue_id
+        })
+        self._queue_in.put(msg)
+
+        if self._stdin:
+            try:
+                while not self._queue_in.empty():
+                    self._stdin.write(self._queue_in.get_nowait())
+            except BrokenPipeError:
+                error_tb(self._vim, 'Crash in child process')
+                error(self._vim, 'stderr=' + str(self._proc.read_error()))
+                self._hnd = None
+        return queue_id
+
+    # def _get(self, queue_id, is_async=False):
+    #     if not self._hnd:
+    #         return []
+    #
+    #     outs = []
+    #     while not self._queue_out.empty():
+    #         outs.append(self._queue_out.get_nowait())
+    #     try:
+    #         return [x for x in outs if x['queue_id'] == queue_id]
+    #     except TypeError:
+    #         error_tb(self._vim,
+    #                  '"stdout" seems contaminated by sources. '
+    #                  '"stdout" is used for RPC; Please pipe or discard')
+    #         return []
+
+    def _get(self, name, args):
+        return None
+
+        if not self._hnd:
+            return
+
+        self._vim.vars['denite#_ret'] = {}
+
+        queue_id = self._put(name, args)
+
+        while queue_id not in self._vim.vars['denite#_ret']:
+            time.sleep(0.1)
+
+        return self._vim.vars['denite#_ret'][queue_id]

rplugin/python3/denite/ui/default.py

@@ -13,7 +13,7 @@
     clear_cmdline, echo, error, regex_convert_py_vim, clearmatch)
 from .action import DEFAULT_ACTION_KEYMAP
 from .prompt import DenitePrompt
-from denite.parent import SyncParent
+from denite.parent import ASyncParent, SyncParent
 from ..prompt.prompt import STATUS_ACCEPT, STATUS_INTERRUPT
 
 
@@ -69,7 +69,8 @@ def __init__(self, vim):
 
     def start(self, sources, context):
         if not self._denite:
-            self._denite = SyncParent(self._vim)
+            # self._denite = SyncParent(self._vim)
+            self._denite = ASyncParent(self._vim)
 
         self._result = []
         context['sources_queue'] = [sources]