refactor so that printing and evaluation result has a deterministic

order at the REPL.

Author: swannodette

src/main/clojure/cljs/repl/node.clj

@@ -21,12 +21,16 @@
            [java.io File Reader BufferedReader BufferedWriter
            InputStreamReader IOException]
            [java.lang ProcessBuilder Process]
-           [java.util.concurrent ConcurrentHashMap]))
+           [java.util.concurrent ConcurrentHashMap LinkedBlockingQueue]))
 
 (def lock (Object.))
+(def results (ConcurrentHashMap.))
 (def outs (ConcurrentHashMap.))
 (def errs (ConcurrentHashMap.))
 
+(defn thread-name []
+  (.getName (Thread/currentThread)))
+
 (defn create-socket [^String host port]
   (let [socket (Socket. host (int port))
         in     (io/reader socket)
@@ -56,14 +60,10 @@
 (defn node-eval
   "Evaluate a JavaScript string in the Node REPL process."
   [repl-env js]
-  (let [{:keys [in out]} @(:socket repl-env)]
-    ;; escape backslash for Node.js under Windows
-    (write out
-      (json/write-str
-        {"repl" (.getName (Thread/currentThread))
-         "form" js}))
-    (let [result (json/read-str
-                   (read-response in) :key-fn keyword)]
+  (let [tname (thread-name)
+        {:keys [out]} @(:socket repl-env)]
+    (write out (json/write-str {:type "eval" :repl tname :form js}))
+    (let [result (.take ^LinkedBlockingQueue (.get results tname))]
       (condp = (:status result)
         "success"
         {:status :success
@@ -88,23 +88,26 @@
 (defn- alive? [proc]
   (try (.exitValue proc) false (catch IllegalThreadStateException _ true)))
 
-(defn- pipe [^Process proc in stream ios]
+(defn- event-loop [^Process proc in]
   ;; we really do want system-default encoding here
-  (with-open [^Reader in (-> in InputStreamReader. BufferedReader.)]
-    (while (alive? proc)
-      (try
-        (let [res (read-response in)]
-          (try
-            (let [{:keys [repl content]} (json/read-str res :key-fn keyword)
-                  stream (or (.get ios repl) stream)]
-              (.write stream content 0 (.length ^String content))
-              (.flush stream))
-            (catch Throwable _
-              (.write stream res 0 (.length res))
-              (.flush stream))))
-        (catch IOException e
-          (when (and (alive? proc) (not (.contains (.getMessage e) "Stream closed")))
-            (.printStackTrace e *err*)))))))
+  (while (alive? proc)
+    (try
+      (let [res (read-response in)]
+        (try
+          (let [{:keys [type repl value] :or {repl "main"} :as event}
+                (json/read-str res :key-fn keyword)]
+            (case type
+              "result"
+              (.offer (.get results repl) event)
+              (when-let [stream (.get (if (= type "out") outs errs) repl)]
+                (.write stream value 0 (.length ^String value))
+                (.flush stream))))
+          (catch Throwable _
+            (.write *out* res 0 (.length res))
+            (.flush *out*))))
+      (catch IOException e
+        (when (and (alive? proc) (not (.contains (.getMessage e) "Stream closed")))
+          (.printStackTrace e *err*))))))
 
 (defn- build-process
   [opts repl-env input-src]
@@ -122,22 +125,19 @@
   ([repl-env] (setup repl-env nil))
   ([{:keys [host port socket state] :as repl-env} opts]
    (let [tname (.getName (Thread/currentThread))]
+     (.put results tname (LinkedBlockingQueue.))
      (.put outs tname *out*)
      (.put errs tname *err*))
    (locking lock
      (when-not @socket
-       (let [out          *out*
-             err          *err*
-             output-dir   (io/file (util/output-directory opts))
+       (let [output-dir   (io/file (util/output-directory opts))
              _            (.mkdirs output-dir)
              of           (io/file output-dir "node_repl.js")
              _            (spit of
                             (string/replace (slurp (io/resource "cljs/repl/node_repl.js"))
                               "var PORT = 5001;"
                               (str "var PORT = " (:port repl-env) ";")))
              proc         (.start (build-process opts repl-env of))
-             _            (do (.start (Thread. (bound-fn [] (pipe proc (.getInputStream proc) out outs))))
-                              (.start (Thread. (bound-fn [] (pipe proc (.getErrorStream proc) err errs)))))
              env          (ana/empty-env)
              core         (io/resource "cljs/core.cljs")
              ;; represent paths as vectors so we can emit JS arrays, this is to
@@ -157,6 +157,7 @@
              (if @socket
                (recur (read-response (:in @socket)))
                (recur nil))))
+         (.start (Thread. (bound-fn [] (event-loop proc (:in @socket)))))
          ;; compile cljs.core & its dependencies, goog/base.js must be available
          ;; for bootstrap to load, use new closure/compile as it can handle
          ;; resources in JARs
@@ -235,7 +236,8 @@
     (load-javascript this provides url))
   (-tear-down [this]
     (swap! state update :listeners dec)
-    (let [tname (Thread/currentThread)]
+    (let [tname (thread-name)]
+      (.remove results tname)
       (.remove outs tname)
       (.remove errs tname))
     (locking lock

src/main/clojure/cljs/repl/node_repl.js

@@ -15,24 +15,6 @@ var dom  = require("domain").create();
 var PORT = 5001;
 var repl = null;
 
-process.stdout.write = (function(write) {
-    return function(chunk, encoding, fd) {
-        var args = Array.prototype.slice.call(arguments, 0);
-        args[0] = JSON.stringify({repl: repl, content: chunk});
-        write.apply(process.stdout, args);
-        write.call(process.stdout, "\0");
-    };
-})(process.stdout.write);
-
-process.stderr.write = (function(write) {
-    return function(chunk, encoding, fd) {
-        var args = Array.prototype.slice.call(arguments, 0);
-        args[0] = JSON.stringify({repl: repl, content: chunk});
-        write.apply(process.stderr, args);
-        write.call(process.stderr, "\0");
-    };
-})(process.stderr.write);
-
 try {
     require("source-map-support").install();
 } catch(err) {
@@ -48,6 +30,23 @@ var server = net.createServer(function (socket) {
 
     socket.setEncoding("utf8");
 
+    process.stdout.write = function(chunk, encoding, fd) {
+        var args = Array.prototype.slice.call(arguments, 0);
+        args[0] = JSON.stringify({type: "out", repl: repl, value: chunk});
+        socket.write.apply(socket, args);
+        socket.write("\0");
+    };
+
+    process.stderr.write = (function(write) {
+        return function(chunk, encoding, fd) {
+            var args = Array.prototype.slice.call(arguments, 0);
+            args[0] = JSON.stringify({type: "err", repl: repl, value: chunk});
+            socket.write.apply(socket, args);
+            socket.write("\0");
+        };
+    })(process.stderr.write);
+
+
     dom.on("error", function(ue) {
         console.error(ue.stack);
     });
@@ -84,18 +83,21 @@ var server = net.createServer(function (socket) {
 
             if(err) {
                 socket.write(JSON.stringify({
+                    type: "result",
                     repl: repl,
                     status: "exception",
                     value: err.stack
                 }));
             } else if(ret !== undefined && ret !== null) {
                 socket.write(JSON.stringify({
+                    type: "result",
                     repl: repl,
                     status: "success",
                     value: ret.toString()
                 }));
             } else {
                 socket.write(JSON.stringify({
+                    type: "result",
                     repl: repl,
                     status: "success",
                     value: null