stream -> InputStream conversion

Author: ztellman

src/byte_streams.clj

@@ -4,6 +4,7 @@
   (:require
     [manifold.stream :as s]
     [manifold.deferred :as d]
+    [byte-streams.pushback-stream :as ps]
     [byte-streams.char-sequence :as cs]
     [byte-streams.utils :refer (fast-memoize)]
     [clojure.java.io :as io]
@@ -27,6 +28,7 @@
      FileOutputStream
      FileInputStream
      ByteArrayInputStream
+     ByteArrayOutputStream
      PipedOutputStream
      PipedInputStream
      DataInputStream
@@ -68,11 +70,7 @@
 (defn- protocol? [x]
   (and (map? x) (contains? x :on-interface)))
 
-(defn seq-of [x]
-  (list 'seq-of x))
-
-(defn stream-of [x]
-  (list 'stream-of x))
+(declare seq-of stream-of)
 
 (defn seq-of? [x]
   (and (seq? x)
@@ -90,6 +88,13 @@
 
 (defn- abstract-type-descriptor [x]
   (cond
+
+    (and (sequential? x) (= 'var (first x)))
+    (abstract-type-descriptor (second x))
+
+    (var? x)
+    x
+
     (seq-of? x)
     (list 'list '(quote seq-of) (abstract-type-descriptor (second x)))
 
@@ -105,6 +110,12 @@
         (:var x)
         x))))
 
+(defn seq-of [x]
+  (list 'seq-of x))
+
+(defn stream-of [x]
+  (list 'stream-of x))
+
 (defmacro def-conversion
   "Defines a conversion from one type to another."
   [[src dst :as conversion] params & body]
@@ -397,9 +408,12 @@
   "Returns a descriptor that can be used with `conversion-path`."
   [x]
   (cond
-    (or (class? x) (protocol? x))
+    (class? x)
     x
 
+    (protocol? x)
+    (:var x)
+
     (contains? @src->dst->conversion (class x))
     (class x)
 
@@ -411,7 +425,7 @@
 
 (defn convert
   "Converts `x`, if possible, into type `dst`, which can be either a class or protocol.  If no such conversion
-   is possible, an IllegalArgumentException is thrown.
+   is possible, an IllegalArgumentException is thrown.  If `x` is a stream, then the `src` type must be explicitly specified.
 
    `options` is a map, whose available settings depend on what sort of transform is being performed:
 
@@ -429,45 +443,17 @@
      (cond
 
        (s/source? x)
-       (if (stream-of? dst)
-
-         ;; -> (stream-of a)
-         (let [s' (s/stream)
-               dst (if (protocol? (second dst))
-                     (:var (second dst))
-                     (second dst))]
-           (-> x
-             (s/take! ::none)
-             (d/chain
-               (fn [msg]
-                 (if (identical? ::none msg)
-                   (s/close! s')
-                   (let [src (type-descriptor msg)]
-                     (if-let [f (converter src dst)]
-                       (do
-                         (s/put! s' (f msg options))
-                         (s/connect-via x #(s/put! s' (f % options)) s'))
-                       (do
-                         (s/close! x)
-                         (s/close! s'))))))))
-           s')
-
-
-         (let [msg @(s/take! x ::none)
-               src (type-descriptor msg)
-               s' (s/stream)]
-           (when-not (identical? ::none msg)
-             (s/put! s' msg)
-             (s/connect x s')
-             (if-let [f (converter (stream-of src) dst)]
-               (f s' options)
-               (throw (IllegalArgumentException. (str "Don't know how to convert a stream of " src " into " dst)))))))
+       (let [src (get options :source-type)]
+         (assert src "must specify `:source-type` when converting streams")
+         (if-let [f (converter src dst)]
+           (f x options)
+           (throw (IllegalArgumentException. (str "don't know how to convert " src " into " dst)))))
 
        (not (or (nil? x) (and (sequential? x) (empty? x))))
-       (let [src (type-descriptor x)
-             dst (if (protocol? dst)
-                   (:var dst)
-                   dst)]
+       (let [src (or
+                   (when (sequential? x)
+                     (get options :source-type))
+                   (type-descriptor x))]
          (if (or
                (= src dst)
                (and (class? src) (class? dst) (.isAssignableFrom ^Class dst src)))
@@ -559,11 +545,11 @@
                 (f source' sink' options))))
 
           (and
-            (conversion-path src ByteSource)
-            (conversion-path dst ByteSink))
+            (conversion-path src #'ByteSource)
+            (conversion-path dst #'ByteSink))
           (fn [source sink {:keys [close?] :or {close? true} :as options}]
-            (let [source' (convert source ByteSource options)
-                  sink' (convert sink ByteSink options)]
+            (let [source' (convert source #'ByteSource options)
+                  sink' (convert sink #'ByteSink options)]
               (default-transfer source' sink' options)
               (when close?
                 (doseq [x [source sink source' sink']]
@@ -593,6 +579,8 @@
   ([source sink]
      (transfer source sink nil))
   ([source sink options]
+     (transfer source nil sink options))
+  ([source source-type sink options]
      (if (s/source? source)
 
        (let [msg @(s/take! source ::none)
@@ -626,6 +614,26 @@
 
 ;;; conversion definitions
 
+(def-conversion ^{:cost 0} [(stream-of byte-array) InputStream]
+  [s options]
+  (let [ps (ps/pushback-stream (get options :buffer-size 65536))]
+    (s/consume
+      (fn [^bytes ary]
+        (ps/put-array ps ary 0 (alength ary)))
+      s)
+    (s/on-drained s #(ps/close ps))
+    (ps/->input-stream ps)))
+
+(def-conversion ^{:cost 0} [(stream-of ByteBuffer) InputStream]
+  [s options]
+  (let [ps (ps/pushback-stream (get options :buffer-size 65536))]
+    (s/consume
+      (fn [buf]
+        (ps/put-buffer ps buf))
+      s)
+    (s/on-drained s #(ps/close ps))
+    (ps/->input-stream ps)))
+
 ;; byte-array => byte-buffer
 (def-conversion ^{:cost 0} [byte-array ByteBuffer]
   [ary {:keys [direct?] :or {direct? false}}]
@@ -747,7 +755,7 @@
           (.close sink))))
     source))
 
-(def-conversion ^{:cost 1.5} [(seq-of ByteSource) InputStream]
+(def-conversion ^{:cost 1.5} [(seq-of #'ByteSource) InputStream]
   [srcs options]
   (let [chunk-size (get options :chunk-size 65536)
         out (PipedOutputStream.)
@@ -762,8 +770,14 @@
           (.close out))))
     in))
 
+(def-conversion ^{:cost 2} [#'ByteSource byte-array]
+  [src options]
+  (let [os (ByteArrayOutputStream.)]
+    (transfer src os)
+    (.toByteArray os)))
+
 ;; generic byte-source => lazy char-sequence
-(def-conversion ^{:cost 2} [ByteSource CharSequence]
+(def-conversion ^{:cost 2} [#'ByteSource CharSequence]
   [source options]
   (cs/decode-byte-source
     #(let [bytes (take-bytes! source % options)]
@@ -1102,14 +1116,14 @@
   ([x]
      (to-byte-source x nil))
   ([x options]
-     (convert x ByteSource options)))
+     (convert x #'ByteSource options)))
 
 (defn to-byte-sink
   "Converts the object to something that satisfies `ByteSink`."
   ([x]
      (to-byte-sink x nil))
   ([x options]
-     (convert x ByteSink options)))
+     (convert x #'ByteSink options)))
 
 ;;;
 
@@ -1119,13 +1133,15 @@
         sign (long (if (pos? diff) -1 1))
         a (if (pos? diff) b' a')
         b (if (pos? diff) a' b')
-        limit (p/>> (.remaining a) 2)]
+        limit (p/>> (.remaining a) 2)
+        a-offset (.position a)
+        b-offset (.position b)]
     (let [cmp (loop [idx 0]
                 (if (p/>= idx limit)
                   0
                   (let [cmp (p/-
-                              (p/int->uint (.getInt a idx))
-                              (p/int->uint (.getInt b idx)))]
+                              (p/int->uint (.getInt a (p/+ idx a-offset)))
+                              (p/int->uint (.getInt b (p/+ idx b-offset))))]
                     (if (p/== 0 cmp)
                       (recur (p/+ idx 4))
                       (p/* sign cmp)))))]
@@ -1135,8 +1151,8 @@
             (if (p/>= idx limit')
               diff
               (let [cmp (p/-
-                          (p/byte->ubyte (.get a idx))
-                          (p/byte->ubyte (.get b idx)))]
+                          (p/byte->ubyte (.get a (p/+ idx a-offset)))
+                          (p/byte->ubyte (.get b (p/+ idx b-offset))))]
                 (if (p/== 0 cmp)
                   (recur (p/inc idx))
                   (p/* sign cmp))))))

src/byte_streams/InputStream.java

@@ -0,0 +1,52 @@
+package byte_streams;
+
+import java.io.IOException;
+
+public class InputStream extends java.io.InputStream {
+
+    public interface Streamable {
+        int available();
+        void close();
+        long skip(long n);
+        int read() throws IOException;
+        int read(byte[] bytes, int offset, int length) throws IOException;
+    }
+
+    private Streamable _s;
+
+    public InputStream(Streamable s) {
+        _s = s;
+    }
+
+    public void close() {
+        _s.close();
+    }
+
+    public int available() {
+        return _s.available();
+    }
+
+    public boolean markSupported() {
+        return false;
+    }
+
+    public void mark(int readlimit) {
+        throw new UnsupportedOperationException();
+    }
+
+    public void reset() {
+        throw new UnsupportedOperationException();
+    }
+
+    public long skip(long n) {
+        return _s.skip(n);
+    }
+
+    public int read() throws IOException {
+        return _s.read();
+    }
+
+    public int read(byte[] bytes, int offset, int length) throws IOException {
+        return _s.read(bytes, offset, length);
+    }
+}

src/byte_streams/char_sequence.clj

@@ -11,6 +11,8 @@
      CharsetDecoder
      CoderResult]))
 
+(set! *unchecked-math* true)
+
 (defn parse-result [^CoderResult result]
   (cond
     (.isUnderflow result) :underflow
@@ -33,38 +35,39 @@
    ^ByteBuffer extra-bytes
    close-fn
    byte-source]
-  (lazy-seq
-    (let [num-bytes (+ (if extra-bytes
-                         (.remaining extra-bytes)
-                         0)
-                      (long chunk-size))
-          len (long
-                (Math/ceil
-                  (/ num-bytes
-                    (.averageCharsPerByte decoder))))
-          out (CharBuffer/allocate len)]
+  (let [chunk-size (long chunk-size)]
+    (lazy-seq
+      (let [num-bytes (+ (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