Handle reset of component state from the repl.

Author: jebberjeb

src/neovim_client/nvim.clj

@@ -1,14 +1,10 @@
 (ns neovim-client.nvim
   (:require [clojure.core.async :as async]
             [clojure.string :as str]
+            [clojure.tools.logging :as log]
             [neovim-client.message :as message :refer [->request-msg]]
-            [neovim-client.rpc :as rpc]))
-
-;; ***** Public *****
-
-(def new rpc/new)
-(def register-method! rpc/register-method!)
-(def stop rpc/stop)
+            [neovim-client.rpc :as rpc])
+  (:import (java.net Socket)))
 
 (defmacro defvim
   [fn-name vim-command & args]
@@ -116,3 +112,26 @@
                                      #(window-get-buffer nvim %))
                                (vim-get-windows nvim))]
       ((set visible-buffers) buffer-name))))
+
+(def register-method! rpc/register-method!)
+(def stop rpc/stop)
+
+(defn new*
+  [input output debug]
+  (let [component (rpc/new input output)]
+    ;; Each time you connect to the same nvim instance using a tcp socket, nvim
+    ;; allocates a new channel.
+    (when debug
+      (let [api-info (vim-get-api-info component)]
+        (vim-command component (format "let g:nvim_tcp_plugin_channel = %s"
+                                       (first api-info)))))
+    component))
+
+(defn new
+  "Connect to msgpack-rpc channel via standard io or TCP socket."
+  ([] (new* System/in System/out) false)
+  ([host port]
+   (log/info "plugin host connecting to nvim at " host ":" port)
+   (let [socket (java.net.Socket. host port)]
+     (.setTcpNoDelay socket true)
+     (new* (.getInputStream socket) (.getOutputStream socket) true))))

src/neovim_client/rpc.clj

@@ -6,8 +6,7 @@
             [neovim-client.message :refer [id value msg-type method params
                                            ->response-msg]
                                    :as msg])
-  (:import [java.io DataInputStream DataOutputStream]
-           [java.net Socket]))
+  (:import (java.io DataInputStream DataOutputStream)))
 
 (defn- method-not-found
   [msg]
@@ -20,7 +19,7 @@
   (let [chan (async/chan 1024)]
     (async/go-loop
       []
-      (when-let [msg (msgpack/unpack (DataInputStream. input-stream))]
+      (when-let [msg (msgpack/unpack input-stream)]
         (log/info "stream -> msg -> in chan: " msg)
         (async/>! chan msg)
         (recur)))
@@ -40,7 +39,7 @@
       []
       (when-let [msg (async/<! chan)]
         (log/info "stream <- msg <- out chan: " msg)
-        (write-msg! (msgpack/pack msg) (DataOutputStream. output-stream))
+        (write-msg! (msgpack/pack msg) output-stream)
         (recur)))
     chan))
 
@@ -68,12 +67,18 @@
   [{:keys [input-stream output-stream out-chan in-chan]}]
   (async/close! out-chan)
   (async/close! in-chan)
+  ;; TODO - drain the out-chan before closing the output-stream.
+  (log/info "closing output stream")
+  (.close output-stream)
+  (log/info "closing input stream")
   (.close input-stream)
-  (.close output-stream))
+  (log/info "input and output streams closed"))
 
-(defn new*
+(defn new
   [input-stream output-stream]
   (let [in-chan (create-input-channel input-stream)
+        input-stream (DataInputStream. input-stream)
+        output-stream (DataOutputStream. output-stream)
         message-table (atom {})
         method-table (atom {})
         component {:input-stream input-stream
@@ -99,11 +104,3 @@
               component (->response-msg (id msg) result) nil)))
         (recur))))
     component))
-
-(defn new
-  "Connect to msgpack-rpc channel via standard io or TCP socket."
-  ([] (new* System/in System/out))
-  ([host port]
-   (let [socket (java.net.Socket. host port)]
-     (.setTcpNoDelay socket true)
-     (new* (.getInputStream socket) (.getOutputStream socket)))))