Poll test

polltest.py

import zmq
import neovim
import threading
from time import sleep

@neovim.plugin
class PollTest(object):
    def __init__(self, vim):
        self.vim = vim
        self.ctx = zmq.Context.instance()
        self.sock = self.ctx.socket(zmq.DEALER)
        self.sock.bind("inproc://sock")
        fd = self.sock.getsockopt(zmq.FD)

        self.stop_poll = self.vim.poll_fd(fd, on_readable=self._on_poll)

    @neovim.command("PollTest")
    def poll_test(self):
        self.on_msg("77")
        t = threading.Thread(target=self._run)
        t.start()

    def _on_poll(self):
        while self.sock.getsockopt(zmq.EVENTS) & zmq.POLLIN:
            self.on_msg(self.sock.recv())

    def on_msg(self, msg):
        self.vim.current.buffer.append(msg)
        if msg == b'5':
            self.stop_poll()

    def _run(self):
        writer = self.ctx.socket(zmq.DEALER)
        writer.connect("inproc://sock")
        for i in range(10):
            sleep(1)
            writer.send_string(str(i))

polltest_greenlet.py

import zmq
import neovim
import threading
from time import sleep

@neovim.plugin
class PollTest(object):
    def __init__(self, vim):
        self.vim = vim
        self.ctx = zmq.Context.instance()
        self.sock = self.ctx.socket(zmq.DEALER)
        self.sock.bind("inproc://sock")
        fd = self.sock.getsockopt(zmq.FD)

        self.stop_poll = self.vim.poll_fd(fd, on_readable=self._on_poll, greenlet=False)

    @neovim.command("PollTest")
    def poll_test(self):
        self.on_msg("77")
        t = threading.Thread(target=self._run)
        t.start()

    def _on_poll(self):
        while self.sock.getsockopt(zmq.EVENTS) & zmq.POLLIN:
            self.vim.async_call(self.on_msg, self.sock.recv())

    def on_msg(self, msg):
        self.vim.current.buffer.append(msg)
        if msg == b'5':
            self.stop_poll()

    def _run(self):
        writer = self.ctx.socket(zmq.DEALER)
        writer.connect("inproc://sock")
        for i in range(10):
            sleep(1)
            writer.send_string(str(i))