Skip to content

Instantly share code, notes, and snippets.

Last active December 31, 2015 00:09
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
feedback streams
(ns test-helpers
(:import [ InputStream OutputStream IOException])
(:require [ :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)
(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 [] []
(close [] (reset! os-closed true))
([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 [] []
(close []
(reset! is-closed true))
([] (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)
(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"))))))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment