testing vim channels

Author: johncf

.gitignore

@@ -0,0 +1,3 @@
+chtest.log
+chtest.err
+server/__pycache__

plugin/chtest.vim

@@ -0,0 +1,18 @@
+function TestStart()
+  let server = expand('<sfile>:p:h') . '/test.py'
+  let err_log = expand('<sfile>:p:h') . '/chtest.err'
+  let cmd = ['python3', server]
+  return job_start(cmd, { 'in_mode': 'json',
+                        \ 'out_mode': 'json',
+                        \ 'out_cb': 'MyHandler',
+                        \ 'err_mode': 'raw',
+                        \ 'err_io': 'file',
+                        \ 'err_name': err_log})
+endfun
+
+func MyHandler(ch, msg)
+  echo "server says: " . a:msg
+endfunc
+
+let g:my_job = TestStart()
+let g:my_ch = job_getchannel(g:my_job)

server.py

@@ -0,0 +1,37 @@
+from __future__ import (absolute_import, division, print_function)
+
+import logging
+from multiprocessing import Queue
+from select import select
+
+__metaclass__ = type  # pylint: disable=invalid-name
+
+
+class Middleman():
+
+    def __init__(self, vimx):
+        self.vimx = vimx
+        self.queue = Queue(maxsize=2)
+        self.logger = logging.getLogger(__name__)
+
+    def safe_exec(self, fn):
+        """ fn should take a VimX object as the only argument. """
+        self.queue.put(['exec', fn])
+
+    def loop(self):
+        x = 1
+        while True:
+            ready, _, _ = select([self.vimx.ch_in, self.queue._reader], [], [], 2)
+            for ev in ready:
+                if ev == self.queue._reader:
+                    req = self.queue.get()
+                    if req[0] == 'exec':
+                        fn = req[1]
+                        fn(self.vimx)
+                    elif req[0] == 'exit':
+                        break
+                elif ev == self.vimx.ch_in:
+                    vi = self.vimx.wait()
+                    self.logger.info("got: %s", vi)
+            self.vimx.send('hello {}'.format(x))
+            x += 1

server/__init__.py

@@ -0,0 +1,95 @@
+import logging
+import json
+
+class VimX:
+
+    def __init__(self, ch_in, ch_out):
+        self.ch_in = ch_in
+        self.ch_out = ch_out
+        self.counter = -1
+        self.buffer = [] # buffer for 'positive' objects
+        self.logger = logging.getLogger(__name__)
+
+    def wait(self, expect=0):
+        """ Blocking function. Use with care!
+            `expect` should either be 0 or a negative number. If `expect == 0`, any positive
+            indexed object is returned. Otherwise, it will queue any positive objects until the
+            first negative object is received. If the received negative object does not match
+            `expect`, then a ValueError is raised.
+        """
+        if expect > 0:
+            raise AssertionError('expect <= 0')
+        if expect == 0 and len(self.buffer) > 0:
+            return self.pop()
+        while True:
+            s = self.ch_in.readline()
+            self.logger.info("read: %s", s)
+            ind, obj = json.loads(s)
+            if (expect == 0 and ind < 0) or (expect < 0 and expect != ind):
+                raise ValueError('Incorrect index received! {} != {}', expect, ind)
+            elif expect < 0 and ind > 0:
+                self.buffer.insert(0, obj)
+            else:
+                break
+        return obj
+
+    def write(self, obj):
+        s = json.dumps(obj)
+        print(s, file=self.ch_out) # line break
+        self.ch_out.flush()
+        self.logger.info("write: %s", s)
+
+    def send(self, obj):
+        self.write([0, obj])
+
+    def call(self, fname, *args, reply=True):
+        obj = ['call', fname, args]
+        if reply:
+            obj += [self.counter]
+            self.counter -= 1
+        self.write(obj)
+        if reply:
+            re = self.wait(expect=self.counter)
+            return re
+
+    def eval(self, expr, reply=True):
+        obj = ['expr', expr]
+        if reply:
+            obj += [self.counter]
+            self.counter -= 1
+        self.write(obj)
+        if reply:
+            re = self.wait(expect=self.counter)
+            return re
+
+    def command(self, cmd):
+        obj = ['ex', expr]
+        self.write(obj)
+
+    def echox(self, msg, level=1):
+        """ Execute echom in vim using appropriate highlighting. """
+        level_map = ['None', 'WarningMsg', 'ErrorMsg']
+        msg = msg.strip().replace('"', '\\"').replace('\n', '\\n')
+        self.command('echohl {} | echom "{}" | echohl None'.format(level_map[level], msg))
+
+    def buffer_add(self, name):
+        """ Create a buffer (if it doesn't exist) and return its number. """
+        bufnr = self.call('bufnr', name, 1)
+        self.call('setbufvar', bufnr, '&bl', 1, reply=False)
+        return bufnr
+
+    def sign_place(self, sign_id, name, bufnr, line):
+        """ Place a sign at the specified location. """
+        self.command("sign place {} name={} line={} buffer={}".format(sign_id, name, line, bufnr))
+
+    def sign_unplace(self, sign_id):
+        """ Hide a sign with specified id. """
+        self.command("sign unplace {}".format(sign_id))
+
+    def get_buffer_name(self, nr):
+        """ Get the buffer name given its number. """
+        return self.call('bufname', nr)
+
+    def abspath(self, relpath):
+        vim_cwd = self.call("getcwd")
+        return path.join(vim_cwd, relpath)

server/vimx.py

@@ -0,0 +1,21 @@
+import logging
+from server.vimx import VimX
+from server import Middleman
+import sys
+
+def main():
+    print('hello from the other side', file=sys.stderr)
+    ch_in = sys.stdin
+    ch_out = sys.stdout
+    vimx = VimX(ch_in, ch_out)
+
+    Middleman(vimx).loop()
+
+if __name__ == '__main__':
+    handler = logging.FileHandler('chtest.log', 'w')
+    handler.formatter = logging.Formatter(
+        '%(asctime)s [%(levelname)s @ '
+        '%(filename)s:%(funcName)s:%(lineno)s] - %(message)s')
+    logging.root.addHandler(handler)
+    logging.root.setLevel(logging.DEBUG)
+    main()