mark 0.2.1-alpha3, fixes #21

Author: ztellman

project.clj

@@ -1,12 +1,12 @@
-(defproject byte-streams "0.2.1-alpha2"
+(defproject byte-streams "0.2.1-alpha3"
   :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.1"]]
-  :profiles {:dev {:dependencies [[org.clojure/clojure "1.7.0"]
-                                  [org.clojure/test.check "0.8.1"]
+                 [manifold "0.1.2-alpha3"]]
+  :profiles {:dev {:dependencies [[org.clojure/clojure "1.8.0-RC5"]
+                                  [org.clojure/test.check "0.9.0"]
                                   [codox-md "0.2.0" :exclusions [org.clojure/clojure]]]}}
   :test-selectors {:stress :stress
                    :default (complement :stress)}

src/byte_streams.clj

@@ -330,8 +330,8 @@
 
 (def-conversion ^{:cost 0} [(stream-of bytes) InputStream]
   [s options]
-  (let [ps (ps/pushback-stream (get options :buffer-size 65536))]
-    (s/consume
+  (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)
@@ -340,8 +340,8 @@
 
 (def-conversion ^{:cost 0} [(stream-of ByteBuffer) InputStream]
   [s options]
-  (let [ps (ps/pushback-stream (get options :buffer-size 65536))]
-    (s/consume
+  (let [ps (ps/pushback-stream (get options :buffer-size 1024))]
+    (s/consume-async
       (fn [^ByteBuffer buf]
         (ps/put-buffer ps (.duplicate buf)))
       s)

src/byte_streams/pushback_stream.clj

@@ -2,6 +2,7 @@
   (:refer-clojure :exclude [take])
   (:require
     [primitive-math :as p]
+    [byte-streams.utils :refer [doit]]
     [manifold
      [utils :as u]
      [stream :as s]
@@ -32,6 +33,10 @@
    deferred
    ^boolean eager?])
 
+(defn trigger [^Consumption c]
+  (let [^ByteBuffer buf (.buf c)]
+    (d/success! (.deferred c) (.position buf))))
+
 (defn put [^ByteBuffer src ^ByteBuffer dst]
   (let [l (.limit src)]
     (.limit src (p/+ (.position src) (p/min (.remaining src) (.remaining dst))))
@@ -121,37 +126,45 @@
    PushbackStream
 
    (put [_ buf]
-     ((either
-        [do]
-        [u/with-lock lock])
+     (let [[consumers d]
+           ((either
+              [do]
+              [u/with-lock* lock])
+
+            (if closed?
+              [nil
+               (d/success-deferred false)]
+
+              [(loop [acc []]
+                 (if-let [^Consumption c (.peek consumers)]
+                   (let [^ByteBuffer out (.buf c)]
+                     (put buf out)
+                     (when (or (.eager? c) (not (.hasRemaining out)))
+                       (.remove consumers)
+                       (recur (conj acc c))))
+                   acc))
 
-       (if closed?
-         (d/success-deferred false)
+               (do
+                 (when (.hasRemaining buf)
+                   (.add buffer buf)
+                   (set! buffer-size (unchecked-int (p/+ buffer-size (.remaining buf)))))
 
-         (do
-           (loop []
-             (when-let [^Consumption c (.peek consumers)]
-               (let [^ByteBuffer out (.buf c)]
-                 (put buf out)
-                 (when (or (.eager? c) (not (.hasRemaining out)))
-                   (.remove consumers)
-                   (d/success! (.deferred c) (.position out))
-                   (recur)))))
+                 (cond
 
-           (when (.hasRemaining buf)
-             (.add buffer buf)
-             (set! buffer-size (unchecked-int (p/+ buffer-size (.remaining buf)))))
+                   deferred
+                   deferred
 
-           (cond
+                   (p/<= buffer-size buffer-capacity)
+                   (d/success-deferred true)
 
-             deferred
-             deferred
+                   :else
+                   (set! deferred (d/deferred))))]))]
 
-             (p/<= buffer-size buffer-capacity)
-             (d/success-deferred true)
+       (when consumers
+         (doit [c consumers]
+           (trigger c)))
 
-             :else
-             (set! deferred (d/deferred)))))))
+       d))
 
    (put [this ary offset length]
      (.put this
@@ -160,21 +173,28 @@
          (.limit (+ offset length)))))
 
    (pushback [_ buf]
-     ((either
-        [do]
-        [u/with-lock lock])
-       (loop []
-         (when-let [^Consumption c (.peek consumers)]
-           (let [^ByteBuffer out (.buf c)]
-             (put buf out)
-             (when (or (.eager? c) (not (.hasRemaining out)))
-               (.remove consumers)
-               (d/success! (.deferred c) (.position out))
-               (recur)))))
-
-       (when (.hasRemaining buf)
-         (.addLast buffer buf)
-         (set! buffer-size (unchecked-int (p/+ buffer-size (.remaining buf)))))))
+     (let [consumers
+           ((either
+              [do]
+              [u/with-lock* lock])
+            (let [consumers
+                  (loop [acc []]
+                    (if-let [^Consumption c (.peek consumers)]
+                      (let [^ByteBuffer out (.buf c)]
+                        (put buf out)
+                        (when (or (.eager? c) (not (.hasRemaining out)))
+                          (.remove consumers)
+                          (recur (conj acc c))))
+                      acc))]
+
+              (when (.hasRemaining buf)
+                (.addLast buffer buf)
+                (set! buffer-size (unchecked-int (p/+ buffer-size (.remaining buf)))))
+
+              consumers))]
+
+       (doit [c consumers]
+         (trigger c))))
 
    (pushback [this ary offset length]
      (.pushback this
@@ -183,47 +203,65 @@
          (.limit (+ offset length)))))
 
    (take [_ ary offset length eager?]
+
      (let [out (-> (ByteBuffer/wrap ary)
                  (.position offset)
                  ^ByteBuffer (.limit (+ offset length))
-                 .slice)]
-       ((either
-          [do]
-          [u/with-lock lock])
-
-        (loop []
-          (when-let [^ByteBuffer in (.peek buffer)]
-            (put in out)
-            (when-not (.hasRemaining in)
-              (.remove buffer))
-            (when (.hasRemaining out)
-              (recur))))
-
-        (set! buffer-size (unchecked-int (p/- buffer-size (.position out))))
-
-        (when (and (p/<= buffer-size buffer-capacity) deferred)
-          (d/success! deferred true)
-          (set! deferred nil))
-
-        (if (or closed?
-              (and (pos? (.position out))
-                (or eager? (not (.hasRemaining out)))))
-          (d/success-deferred (.position out))
-          (let [d (d/deferred)]
-            (.add consumers (Consumption. out d eager?))
-            d)))))
+                 .slice)
+
+           [put take]
+
+           ((either
+              [do]
+              [u/with-lock* lock])
+
+            (loop []
+              (when-let [^ByteBuffer in (.peek buffer)]
+                (put in out)
+                (when-not (.hasRemaining in)
+                  (.remove buffer))
+                (when (.hasRemaining out)
+                  (recur))))
+
+            (set! buffer-size
+              (unchecked-int
+                (p/- buffer-size
+                  (p/-
+                    (.position out)
+                    offset))))
+
+            [(when (and (p/<= buffer-size buffer-capacity) deferred)
+               (let [d deferred]
+                 (set! deferred nil)
+                 d))
+
+             (if (or closed?
+                   (and (pos? (.position out))
+                     (or eager? (not (.hasRemaining out)))))
+               (d/success-deferred (.position out))
+               (let [d (d/deferred)]
+                 (.add consumers (Consumption. out d eager?))
+                 d))])]
+
+       (when put
+         (d/success! put true))
+
+       take))
 
    (close [_]
-     ((either
-        [do]
-        [u/with-lock lock])
-       (set! closed? true)
+     (when ((either
+              [do]
+              [u/with-lock* lock])
+            (when-not closed?
+              (set! closed? true)
+              true))
        (loop []
          (when-let [^Consumption c (.poll consumers)]
            (let [^ByteBuffer buf (.buf c)]
              (d/success! (.deferred c) (.position buf)))
-           (recur)))
-       true))))
+           (recur))))
+
+     true)))
 
 (defn pushback-stream [capacity]
   (SynchronizedPushbackByteStream.

src/byte_streams/utils.clj

@@ -5,6 +5,18 @@
     [java.util.concurrent
      ConcurrentHashMap]))
 
+(defmacro doit
+  "A version of doseq that doesn't emit all that inline-destroying chunked-seq code."
+  [[x it] & body]
+  (let [it-sym (gensym "iterable")]
+    `(let [~it-sym ~it
+           it# (.iterator ~(with-meta it-sym {:tag "Iterable"}))]
+       (loop []
+         (when (.hasNext it#)
+           (let [~x (.next it#)]
+            ~@body)
+           (recur))))))
+
 (defmacro memoize-form [m f & args]
   `(let [k# (t/vector ~@args)]
      (let [v# (.get ~m k#)]

test/byte_streams_test.clj

@@ -37,6 +37,7 @@
                                distinct
                                (map (partial map eval')))]
     (doseq [[src dst] pairwise-conversions]
+
       (is (= text
             (-> text
               (convert src)
@@ -59,6 +60,7 @@
                                distinct
                                (map (partial map eval')))]
     (doseq [[src dst] pairwise-conversions]
+
       (is (= (seq ary)
             (-> ary
               (convert src)
@@ -85,6 +87,7 @@
   (doseq [dst (->> String
                 possible-conversions
                 (map eval'))]
+
     (let [file (temp-file)
           file' (temp-file)]
       (transfer (convert text dst) dev-null)