feedback streams

test_helpers.clj

(ns test-helpers
  (:import [java.io InputStream OutputStream IOException])
  (:require [clojure.java.io :as io]))

(defn- swap-!
  "Atomically swaps the value of atom to be:
  (apply f current-value-of-atom args). Note that f may be called
  multiple times, and thus should be free of side effects.  Returns
  the value prior to the update."
  [a f & args]
  (loop []
    (let [oldval @a
          newval (apply f oldval args)]
      (if (compare-and-set! a oldval newval)
        oldval
        (recur)))))

(defn feedback
  "Return a [output input] pair of streams where the output stream feeds back into the input stream"
  []
  (let [q (atom (clojure.lang.PersistentQueue/EMPTY))
        os-closed (atom false)
        is-closed (atom false)
        os (proxy [java.io.OutputStream] []
             (close [] (reset! os-closed true))
             (write
               ([b] (do ;; Overloaded by type
                      (if (instance? (Class/forName "[B") b)
                        (.write ^OutputStream this b 0 (alength ^bytes b))
                        (if @os-closed
                          (throw (IOException. "Stream is closed"))
                          (swap! q conj (byte (.byteValue ^Integer b)))))))
               ([bs offset length] (doseq [i (range length)]
                                     (.write ^OutputStream this (int (aget ^bytes bs i)))))))
        is (proxy [java.io.InputStream] []
             (close []
               (reset! is-closed true))
             (read
               ([] (if @is-closed
                     (throw (IOException. "Stream is closed"))
                     (if-let [v (peek (swap-! q pop))]
                       (int (bit-and 0xff v))
                       (int -1))))
               ([bs] (.read ^InputStream this bs (int 0) (alength ^bytes bs)))
               ([bs off len] (loop [i off l 0]
                               (if (= l len)
                                 len
                                 (let [v (.read ^InputStream this)]
                                   (if (= (int -1) v)
                                     (if (pos? l) l v)
                                     (do (aset-byte bs i (byte (.byteValue v)))
                                         (recur (inc i) (inc l))))))))))]
    [os is]))

(comment (binding [*warn-on-reflection* true]
           (let [s "⠓⠑⠇⠇⠕MyText⠓⠑⠇⠇⠕"
                 bs (.getBytes s)
                 [os is] (feedback)]
             (io/copy bs os)
             (assert (= s (slurp is :encoding "UTF8"))))))