conversions to CharSequences should be via ByteBuffers, not byte-arrays

Author: ztellman

project.clj

@@ -1,11 +1,11 @@
-(defproject byte-streams "0.2.0"
+(defproject byte-streams "0.2.1-SNAPSHOT"
   :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.1"]
+                 [clj-tuple "0.2.2"]
                  [manifold "0.1.0"]]
-  :profiles {:dev {:dependencies [[org.clojure/clojure "1.7.0-alpha6"]
+  :profiles {:dev {:dependencies [[org.clojure/clojure "1.7.0"]
                                   [org.clojure/test.check "0.7.0"]
                                   [codox-md "0.2.0" :exclusions [org.clojure/clojure]]]}}
   :test-selectors {:stress :stress

src/byte_streams.clj

@@ -431,10 +431,9 @@
 ;; channel => lazy-seq of byte-buffers
 (def-conversion [ReadableByteChannel (seq-of ByteBuffer)]
   [channel {:keys [chunk-size direct?] :or {chunk-size 4096, direct? false} :as options}]
-  (when (.isOpen channel)
-    (lazy-seq
-      (when-let [b (proto/take-bytes! channel chunk-size options)]
-        (cons b (convert channel (seq-of ByteBuffer) options))))))
+  (lazy-seq
+    (when-let [b (proto/take-bytes! channel chunk-size options)]
+      (cons b (convert channel (seq-of ByteBuffer) options)))))
 
 ;; input-stream => channel
 (def-conversion ^{:cost 0} [InputStream ReadableByteChannel]
@@ -511,7 +510,7 @@
   [source options]
   (cs/decode-byte-source
     #(when-let [bytes (proto/take-bytes! source % options)]
-       (convert bytes byte-array options))
+       (convert bytes ByteBuffer options))
     #(when (proto/closeable? source)
        (proto/close source))
     options))
@@ -676,17 +675,16 @@
 
   ReadableByteChannel
   (take-bytes! [this n {:keys [direct?] :or {direct? false}}]
-    (when (.isOpen this)
-      (let [^ByteBuffer buf (if direct?
-                              (ByteBuffer/allocateDirect n)
-                              (ByteBuffer/allocate n))]
-        (while
-          (and
-            (.isOpen this)
-            (pos? (.read this buf))))
-
-        (when (pos? (.position buf))
-          (.flip buf)))))
+    (let [^ByteBuffer buf (if direct?
+                            (ByteBuffer/allocateDirect n)
+                            (ByteBuffer/allocate n))]
+      (while
+        (and
+          (.isOpen this)
+          (pos? (.read this buf))))
+
+      (when (pos? (.position buf))
+        (.flip buf))))
 
   ByteBuffer
   (take-bytes! [this n _]

src/byte_streams/char_sequence.clj

@@ -44,40 +44,39 @@
    ^ByteBuffer extra-bytes
    close-fn
    byte-source]
-  (let [chunk-size (long chunk-size)]
-    (lazy-seq
-      (let [num-bytes (+ (long
-                           (if extra-bytes
-                             (.remaining extra-bytes)
-                             0))
-                        (long chunk-size))
-            len (long
-                  (Math/ceil
-                    (/ num-bytes
-                      (.averageCharsPerByte decoder))))
-            out (CharBuffer/allocate len)]
+  (lazy-seq
+    (let [num-bytes (+ (long
+                         (if extra-bytes
+                           (.remaining extra-bytes)
+                           0))
+                      (long chunk-size))
+          len (long
+                (Math/ceil
+                  (/ num-bytes
+                    (.averageCharsPerByte decoder))))
+          out (CharBuffer/allocate len)]
 
-        (if (and extra-bytes (= :overflow (decode decoder extra-bytes out)))
+      (if (and extra-bytes (= :overflow (decode decoder extra-bytes out)))
 
-          ;; we didn't even exhaust the overflow bytes, try again
-          (cons
-            out
-            (lazy-char-buffer-sequence decoder chunk-size extra-bytes close-fn byte-source))
+        ;; we didn't even exhaust the overflow bytes, try again
+        (cons
+          out
+          (lazy-char-buffer-sequence decoder chunk-size extra-bytes close-fn byte-source))
 
-          (if-let [in (byte-source chunk-size)]
-            (let [result (decode decoder in out)]
-              (cons
-                (.flip out)
-                (lazy-char-buffer-sequence
-                  decoder
-                  chunk-size
-                  (when (= :overflow result) in)
-                  close-fn
-                  byte-source)))
-            (do
-              (flush decoder out)
-              (when close-fn (close-fn))
-              (.flip out))))))))
+        (if-let [in (byte-source chunk-size)]
+          (let [result (decode decoder in out)]
+            (cons
+              (.flip out)
+              (lazy-char-buffer-sequence
+                decoder
+                chunk-size
+                (when (= :overflow result) in)
+                close-fn
+                byte-source)))
+          (do
+            (flush decoder out)
+            (when close-fn (close-fn))
+            (.flip out)))))))
 
 (defn decode-byte-source
   [byte-source
@@ -90,9 +89,7 @@
         decoder (doto (.newDecoder (Charset/forName encoding))
                   (.onMalformedInput action)
                   (.onUnmappableCharacter action))
-        s (lazy-char-buffer-sequence decoder chunk-size nil close-fn
-            #(when-let [ary (byte-source %)]
-               (ByteBuffer/wrap ary)))]
+        s (lazy-char-buffer-sequence decoder chunk-size nil close-fn byte-source)]
     (reify
       java.io.Closeable
       (close [_] (when close-fn (close-fn)))

src/byte_streams/graph.clj

@@ -127,7 +127,7 @@
                    (assoc-in [(Type. 'seq src) (Type. 'seq dst)]
                      (Conversion. (fn [x options] (map #(f % options) x)) cost))
                    (assoc-in [(Type. 'stream src) (Type. 'stream dst)]
-                     (Conversion. (fn [x options] (s/map #(f % options) x)) cost))))
+                     (Conversion. (fn [x options] (s/map #(f % options) x)) (+ cost 0.1)))))
                m')]
       (ConversionGraph. m')))
   (possible-sources [_]