first cut of prepl

src/clj/clojure/core.clj

@@ -3759,6 +3759,21 @@
   ([opts stream]
    (. clojure.lang.LispReader (read stream opts))))
 
+(defn read+string
+  "Like read, and taking the same args. stream must be a LineNumberingPushbackReader.
+  Returns a vector containing the object read and the (whitespace-trimmed) string read."
+  {:added "1.10"}
+  ([] (read+string *out*))
+  ([^clojure.lang.LineNumberingPushbackReader stream & args]
+     (try
+       (.captureString stream)
+       (let [o (apply read stream args)
+             s (.trim (.getString stream))]
+         [o s])       
+       (catch Throwable ex
+         (.getString stream)
+         (throw ex)))))
+
 (defn read-line
   "Reads the next line from stream that is the current value of *in* ."
   {:added "1.0"
@@ -7766,3 +7781,43 @@
   "Return true if x is a java.net.URI"
   {:added "1.9"}
   [x] (instance? java.net.URI x))
+
+(defonce ^:private tapset (atom #{}))
+(defonce ^:private ^java.util.concurrent.ArrayBlockingQueue tapq (java.util.concurrent.ArrayBlockingQueue. 1024))
+
+(defn add-tap
+  "adds f, a fn of one argument, to the tap set. This function will be called with anything sent via tap>.
+  This function may (briefly) block (e.g. for streams), and will never impede calls to tap>,
+  but blocking indefinitely may cause tap values to be dropped.
+  Remember f in order to remove-tap"
+  {:added "1.10"}
+  [f]
+  (swap! tapset conj f)
+  nil)
+
+(defn remove-tap
+  "remove f from the tap set the tap set."
+  {:added "1.10"}
+  [f]
+  (swap! tapset disj f)
+  nil)
+
+(defn tap>
+  "sends x to any taps. Will not block. Returns true if there was room in the queue,
+  false if not (dropped)."
+  {:added "1.10"}
+  [x]
+  (.offer tapq x))
+
+(defonce ^:private tap-loop
+  (doto (Thread.
+         #(let [x (.take tapq)
+                taps @tapset]
+            (doseq [tap taps]
+              (try
+                (tap x)
+                (catch Throwable ex)))
+            (recur))
+         "clojure.core/tap-loop")
+    (.setDaemon true)
+    (.start)))

src/clj/clojure/core/server.clj

@@ -12,8 +12,11 @@
   (:require [clojure.string :as str]
             [clojure.edn :as edn]
             [clojure.main :as m])
-  (:import [java.net InetAddress Socket ServerSocket SocketException]
-           [java.util.concurrent.locks ReentrantLock]))
+  (:import
+   [clojure.lang LineNumberingPushbackReader]
+   [java.net InetAddress Socket ServerSocket SocketException]
+   [java.io Reader Writer PrintWriter BufferedWriter BufferedReader InputStreamReader OutputStreamWriter]
+   [java.util.concurrent.locks ReentrantLock]))
 
 (set! *warn-on-reflection* true)
 
@@ -106,8 +109,8 @@
           (when (not (.isClosed socket))
             (try
               (let [conn (.accept socket)
-                    in (clojure.lang.LineNumberingPushbackReader. (java.io.InputStreamReader. (.getInputStream conn)))
-                    out (java.io.BufferedWriter. (java.io.OutputStreamWriter. (.getOutputStream conn)))
+                    in (LineNumberingPushbackReader. (InputStreamReader. (.getInputStream conn)))
+                    out (BufferedWriter. (OutputStreamWriter. (.getOutputStream conn)))
                     client-id (str client-counter)]
                 (thread
                   (str "Clojure Connection " name " " client-id) client-daemon
@@ -179,4 +182,130 @@
   []
   (m/repl
     :init repl-init
-    :read repl-read))
+    :read repl-read))
+
+(defn prepl
+  "a REPL with structured output (for programs)
+  reads forms to eval from in-reader (a LineNumberingPushbackReader)
+  Closing the input or passing the form :repl/quit will cause it to return
+
+  Calls out-fn with data, one of:
+  {:tag :ret
+   :val val ;;eval result
+   :ns ns-name-string
+   :ms long ;;eval time in milliseconds
+   :form string ;;iff successfully read
+  }
+  {:tag :out
+   :val string} ;chars from during-eval *out*
+  {:tag :err
+   :val string} ;chars from during-eval *err*
+  {:tag :tap
+   :val val} ;values from tap>
+
+  You might get more than one :out or :err per eval, but exactly one :ret
+  tap output can happen at any time (i.e. between evals)
+  If during eval an attempt is made to read *in* it will read from in-reader unless :stdin is supplied
+"
+  [in-reader out-fn & {:keys [stdin]}]
+  (let [EOF (Object.)
+        tapfn #(out-fn {:tag :tap :val %1})]
+    (m/with-bindings
+      (in-ns 'user)
+      (binding [*in* (or stdin in-reader)
+                *out* (PrintWriter-on #(out-fn {:tag :out :val %1}) nil)
+                *err* (PrintWriter-on #(out-fn {:tag :err :val %1}) nil)]
+        (try
+          (add-tap tapfn)
+          (loop []
+            (when (try
+                    (let [[form s] (read+string in-reader false EOF)]
+                      (try
+                        (when-not (identical? form EOF)
+                          (let [start (System/nanoTime)
+                                ret (eval form)
+                                ms (quot (- (System/nanoTime) start) 1000000)]
+                            (when-not (= :repl/quit ret)
+                              (set! *3 *2)
+                              (set! *2 *1)
+                              (set! *1 ret)
+                              (out-fn {:tag :ret
+                                       :val (if (instance? Throwable ret)
+                                              (Throwable->map ret)
+                                              ret)
+                                       :ns (str (.name *ns*))
+                                       :ms ms
+                                       :form s})
+                              true)))
+                        (catch Throwable ex
+                          (set! *e ex)
+                          (out-fn {:tag :ret :val (Throwable->map ex) :ns (str (.name *ns*)) :form s})
+                          true)))
+                    (catch Throwable ex
+                      (set! *e ex)
+                      (out-fn {:tag :ret :val (Throwable->map ex) :ns (str (.name *ns*))})
+                      true))
+              (recur)))
+          (finally
+           (remove-tap tapfn)))))))
+
+(defn- resolve-fn [valf]
+  (if (symbol? valf)
+    (or (resolve valf)
+        (when-let [nsname (namespace valf)]
+          (require (symbol nsname))
+          (resolve valf))
+        (throw (Exception. (str "can't resolve: " valf))))
+    valf))
+
+(defn io-prepl
+  "prepl bound to *in* and *out*, suitable for use with e.g. server/repl (socket-repl).
+  :ret and :tap vals will be processed by valf, a fn of one argument
+  or a symbol naming same (default pr-str)"
+  [& {:keys [valf] :or {valf pr-str}}]
+  (let [valf (resolve-fn valf)
+        out *out*
+        lock (Object.)]
+    (prepl *in*
+           #(binding [*out* out, *flush-on-newline* true, *print-readably* true]
+              (locking lock
+                (prn (cond-> %1
+                             (#{:ret :tap} (:tag %1))
+                             (assoc :val (valf (:val %1))))))))))
+
+(defn remote-prepl
+  "Implements a prepl on in-reader and out-fn by forwarding to a
+  remote [io-]prepl over a socket.  Messages will be read by readf, a
+  fn of a LineNumberingPushbackReader and EOF value or a symbol naming
+  same (default #(read %1 false %2)),
+  :ret and :tap vals will be processed by valf, a fn of one argument
+  or a symbol naming same (default read-string). If that function
+  throws, :val will be unprocessed."
+  [^String host port ^Reader
+  in-reader out-fn & {:keys [valf readf] :or {valf read-string, readf #(read %1 false %2)}}]
+  (let [valf (resolve-fn valf)
+        readf (resolve-fn readf)
+        ^long port (if (string? port) (Integer/valueOf ^String port) port)
+        socket (Socket. host port)
+        rd (-> socket .getInputStream InputStreamReader. BufferedReader. LineNumberingPushbackReader.)
+        wr (-> socket .getOutputStream OutputStreamWriter.)
+        EOF (Object.)]
+    (thread "clojure.core.server/remote-prepl" true
+            (try (loop []
+                   (let [{:keys [tag val] :as m} (readf rd EOF)]
+                     (when-not (identical? m EOF)
+                       (out-fn (cond-> m
+                                       (#{:ret :tap} tag)
+                                       (assoc :val (try (valf val) (catch Throwable ex val)))))
+                       (recur))))
+                 (finally
+                  (.close wr))))
+    (let [buf (char-array 1024)]
+      (try (loop []
+             (let [n (.read in-reader buf)]
+               (when-not (= n -1)
+                 (.write wr buf 0 n)
+                 (.flush wr)
+                 (recur))))
+           (finally
+            (.close rd))))))

src/clj/clojure/core_print.clj

@@ -545,3 +545,27 @@
   (print-method (:form o) w))
 
 (def ^{:private true} print-initialized true)
+
+(defn ^java.io.PrintWriter PrintWriter-on
+  "implements java.io.PrintWriter given flush-fn, which will be called
+  when .flush() is called, with a string built up since the last call to .flush().
+  if not nil, close-fn will be called with no arguments when .close is called"
+  {:added "1.10"}
+  [flush-fn close-fn]
+  (let [sb (StringBuilder.)]
+    (-> (proxy [Writer] []
+          (flush []
+                 (when (pos? (.length sb))
+                   (flush-fn (.toString sb)))
+                 (.setLength sb 0))
+          (close []
+                 (.flush ^Writer this)
+                 (when close-fn (close-fn))
+                 nil)
+          (write [str-cbuf off len]
+                 (when (pos? len)
+                   (if (instance? String str-cbuf)
+                     (.append sb ^String str-cbuf ^int off ^int len)
+                     (.append sb ^chars str-cbuf ^int off ^int len)))))
+        java.io.BufferedWriter.
+        java.io.PrintWriter.)))

src/jvm/clojure/lang/LineNumberingPushbackReader.java

@@ -75,7 +75,7 @@ public int read() throws IOException{
         _atLineStart = false;
         _columnNumber++;
         }
-    if(sb != null)
+    if(sb != null && c != -1)
         sb.append((char)c);
     return c;
 }