consume-async wasn't quite what I needed, mark 0.2.1-alpha4

Author: Zach Tellman

project.clj

@@ -1,11 +1,11 @@
-(defproject byte-streams "0.2.1-alpha3"
+(defproject byte-streams "0.2.1-alpha4"
   :description "A simple way to handle the menagerie of Java byte represenations."
   :license {:name "MIT License"
             :url "http://opensource.org/licenses/MIT"}
   :dependencies [[primitive-math "0.1.4"]
                  [clj-tuple "0.2.2"]
-                 [manifold "0.1.2-alpha3"]]
-  :profiles {:dev {:dependencies [[org.clojure/clojure "1.8.0-RC5"]
+                 [manifold "0.1.2"]]
+  :profiles {:dev {:dependencies [[org.clojure/clojure "1.8.0"]
                                   [org.clojure/test.check "0.9.0"]
                                   [codox-md "0.2.0" :exclusions [org.clojure/clojure]]]}}
   :test-selectors {:stress :stress

src/byte_streams.clj

@@ -54,6 +54,7 @@
 ;;;
 
 (defonce conversions (atom (g/conversion-graph)))
+(defonce inverse-conversions (atom (g/conversion-graph)))
 (defonce src->dst->transfer (atom nil))
 
 (def ^:private ^:const byte-array (class (Utils/byteArray 0)))
@@ -106,17 +107,21 @@
   [[src dst :as conversion] params & body]
   (let [^Type src (normalize-type-descriptor src)
         dst (normalize-type-descriptor dst)]
-    `(swap! conversions g/assoc-conversion ~src ~dst
-       (fn [~(with-meta (first params)
-               {:tag (when (and (instance? Class (.type src)) (not (.wrapper src)))
-                       (if (= src (normalize-type-descriptor 'bytes))
-                         'bytes
-                         (.getName ^Class (.type src))))})
-            ~(if-let [options (second params)]
-               options
-               `_#)]
-         ~@body)
-       ~(get (meta conversion) :cost 1))))
+    `(let [f#
+           (fn [~(with-meta (first params)
+                   {:tag (when (and (instance? Class (.type src)) (not (.wrapper src)))
+                           (if (= src (normalize-type-descriptor 'bytes))
+                             'bytes
+                             (.getName ^Class (.type src))))})
+                ~(if-let [options (second params)]
+                   options
+                   `_#)]
+             ~@body)
+
+           cost#
+           ~(get (meta conversion) :cost 1)]
+       (swap! conversions g/assoc-conversion ~src ~dst f# cost#)
+       (swap! inverse-conversions g/assoc-conversion ~dst ~src f# cost#))))
 
 (defmacro def-transfer
   "Defines a byte transfer from one type to another."
@@ -331,21 +336,28 @@
 (def-conversion ^{:cost 0} [(stream-of bytes) InputStream]
   [s options]
   (let [ps (ps/pushback-stream (get options :buffer-size 1024))]
-    (s/consume-async
-      (fn [^bytes ary]
-        (ps/put-array ps ary 0 (alength ary)))
-      s)
-    (s/on-drained s #(ps/close ps))
+    (d/loop []
+      (d/chain (s/take! s ::none)
+        (fn [^bytes msg]
+          (if (identical? ::none msg)
+            (ps/close ps)
+            (
+             do
+              (ps/put-array ps msg 0 (alength msg))
+              (d/recur))))))
     (ps/->input-stream ps)))
 
 (def-conversion ^{:cost 0} [(stream-of ByteBuffer) InputStream]
   [s options]
   (let [ps (ps/pushback-stream (get options :buffer-size 1024))]
-    (s/consume-async
-      (fn [^ByteBuffer buf]
-        (ps/put-buffer ps (.duplicate buf)))
-      s)
-    (s/on-drained s #(ps/close ps))
+    (d/loop []
+      (d/chain (s/take! s ::none)
+        (fn [^ByteBuffer msg]
+          (if (identical? ::none msg)
+            (ps/close ps)
+            (do
+              (ps/put-buffer ps (.duplicate msg))
+              (d/recur))))))
     (ps/->input-stream ps)))
 
 ;; byte-array => byte-buffer

src/byte_streams/pushback_stream.clj

@@ -91,15 +91,9 @@
     [lock
      ^LinkedList consumers
      ^long buffer-capacity
-     (either
-       [^:unsynchronized-mutable ^int buffer-size]
-       [^:volatile-mutable ^int buffer-size])
-     (either
-       [^:unsynchronized-mutable deferred]
-       [^:volatile-mutable deferred])
-     (either
-       [^:unsynchronized-mutable closed?]
-       [^:volatile-mutable closed?])
+     ^:unsynchronized-mutable ^int buffer-size
+     ^:unsynchronized-mutable deferred
+     ^:unsynchronized-mutable closed?
      ^LinkedList buffer]
 
     InputStream$Streamable
@@ -126,6 +120,7 @@
    PushbackStream
 
    (put [_ buf]
+
      (let [[consumers d]
            ((either
               [do]
@@ -223,12 +218,7 @@
                 (when (.hasRemaining out)
                   (recur))))
 
-            (set! buffer-size
-              (unchecked-int
-                (p/- buffer-size
-                  (p/-
-                    (.position out)
-                    offset))))
+            (set! buffer-size (unchecked-int (p/- buffer-size (.position out))))
 
             [(when (and (p/<= buffer-size buffer-capacity) deferred)
                (let [d deferred]