byte_streams.clj

(ns clj-commons.byte-streams
  (:refer-clojure :exclude [byte-array vector-of])
  (:require
   [manifold
    [stream :as s]
    [deferred :as d]]
   [clj-commons.byte-streams
    [graph :as g]
    [protocols :as proto]
    [pushback-stream :as ps]
    [char-sequence :as cs]]
   [clj-commons.primitive-math :as p])
  (:import
   (clj_commons.byte_streams
    Utils
    ByteBufferInputStream)
   (clj_commons.byte_streams.graph
    Type)
   (java.nio
    ByteBuffer)
   (java.lang.reflect
    Array)
   (java.io
    File
    ByteArrayInputStream
    ByteArrayOutputStream
    PipedOutputStream
    PipedInputStream
    DataInputStream
    InputStream
    OutputStream
    IOException
    Reader
    BufferedReader
    InputStreamReader)
   (java.nio.channels
    ReadableByteChannel
    WritableByteChannel
    FileChannel
    FileChannel$MapMode
    Channels
    Pipe)
   (java.nio.channels.spi
    AbstractSelectableChannel)
   (java.nio.file StandardOpenOption)))

;;;

(defonce conversions (atom (g/conversion-graph)))
(defonce inverse-conversions (atom (g/conversion-graph)))
(defonce src->dst->transfer (atom nil))

(def ^:private ^:const byte-array-type (class (Utils/byteArray 0)))

(defn seq-of [x]
  (g/type 'seq (if (identical? bytes x) byte-array-type x)))

(defn stream-of [x]
  (g/type 'stream (if (identical? bytes x) byte-array-type x)))

(defn vector-of [x]
  (g/type 'vector (if (identical? bytes x) byte-array-type x)))

(defn type-descriptor
  "Returns a descriptor of the type of the given instance."
  [x]
  (cond

    (nil? x)
    (g/type ::nil)

    (identical? bytes x)
    (g/type byte-array-type)

    (vector? x)
    (vector-of (.type ^Type (type-descriptor (first x))))

    (sequential? x)
    (seq-of nil)

    (s/source? x)
    (stream-of nil)

    :else
    (g/type (class x))))

(defn- normalize-type-descriptor [x]
  (cond
    (instance? Type x)
    x

    (or (= 'bytes x) (= bytes x))
    (g/type byte-array-type)

    :else
    (g/type (eval x))))

(defn- tag-metadata-for [^Type src]
  (if (and (instance? Class (.type src))
           (not (.wrapper src)))
    {:tag (if (= src (normalize-type-descriptor 'bytes))
            'bytes
            (.getName ^Class (.type src)))}
    {}))

(defmacro def-conversion
  "Defines a conversion from one type to another."
  [[src dst :as conversion] params & body]
  (let [^Type src (normalize-type-descriptor src)
        dst (normalize-type-descriptor dst)]
    `(let [f#
           (fn [~(with-meta (first params)
                   (tag-metadata-for 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."
  [[src dst] params & body]
  (let [src (normalize-type-descriptor src)
        dst (normalize-type-descriptor dst)
        src-meta (tag-metadata-for src)
        dst-meta (tag-metadata-for dst)]
    `(swap! src->dst->transfer assoc-in [~src ~dst]
            (fn [~(with-meta (first params) src-meta)
                 ~(with-meta (second params) dst-meta)
                 ~(if-let [options (get params 2)] options (gensym "options"))]
              ~@body))))

;;; convert

(def ^:private converter
  (memoize
   (fn [src dst]
     (g/conversion-fn @conversions src dst))))

(declare convert)

(def ^:private seq-converter
  (memoize
   (fn [dst]
     (g/seq-conversion-fn @conversions convert 'seq dst))))

(def ^:private stream-converter
  (memoize
   (fn [dst]
     (g/seq-conversion-fn @conversions convert 'stream dst))))

(defn conversion-path [src dst]
  (let [path (-> @conversions
                 (g/conversion-path (g/type src) (g/type dst))
                 :path)]
    (map (partial mapv g/pprint-type) path)))

(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.  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:

   `chunk-size`  - if a stream is being transformed into a sequence of discrete chunks, `:chunk-size` describes the
                   size of the chunks, which default to 4096 bytes.

   `encoding`    - if a string is being encoded or decoded, `:encoding` describes the charset that is used, which
                   defaults to 'UTF-8'

   `direct?`     - if a byte-buffer is being allocated, `:direct?` describes whether it should be a direct buffer,
                   defaulting to false
   `source-type` - overrides input type detection, required to convert a stream

   (NB: if you need to convert a stream to a seq, or vice versa, but not the underlying byte type, you want
   Manifold's `stream->seq` and `->source` instead)"
  ([x dst]
   (convert x dst nil))
  ([x dst options]
   (let [dst (g/type dst)
         source-type (get options :source-type)
         ^Type
         src (g/type
              (or source-type
                  (type-descriptor x)))
         wrapper (.wrapper src)]
     (cond

       (not (nil? (.type src)))
       (if-let [f (or
                   (converter src dst)
                   (converter (g/type (class x)) dst))]
         (f x (if source-type (dissoc options :source-type) options))
         (throw
          (IllegalArgumentException.
           (str "Don't know how to convert " (class x) " into " (g/pprint-type dst)))))

       (= 'seq wrapper)
       (if-let [f (seq-converter dst)]
         (f x (if source-type (dissoc options :source-type) options))
         x)

       (= 'stream wrapper)
       (if-let [f (stream-converter dst)]
         (f x (if source-type (dissoc options :source-type) options))
         x)

       :else
       (throw (IllegalArgumentException. (str "invalid wrapper type: " (pr-str wrapper) " " (pr-str (.type src)))))))))

(defn possible-conversions
  "Returns a list of all possible conversion targets from value."
  [src]
  (let [^Type src (g/type src)
        pred (cond
               (.type src)
               (partial converter src)

               (= 'seq (.wrapper src))
               seq-converter

               (= 'stream (.wrapper src))
               stream-converter)]
    (->> @conversions
         g/possible-targets
         (filter pred)
         (map g/pprint-type))))

(let [memoized-cost (memoize
                     (fn [src dst]
                       (if-let [path (g/conversion-path @conversions src dst)]
                         (:cost path)
                         9999)))]
  (defn conversion-cost
    "Returns the estimated cost of converting the data `x` to the destination type `dst`."
    ^long [x dst]
    (memoized-cost (type-descriptor x) (normalize-type-descriptor dst))))

;;; transfer

(defn- default-transfer
  [source sink {:keys [chunk-size] :or {chunk-size 1024} :as options}]
  (loop []
    (when-let [b (proto/take-bytes! source chunk-size options)]
      (proto/send-bytes! sink b options)
      (recur))))

(def ^:private transfer-fn
  (memoize
   (fn this [^Type src ^Type dst]
     (let [converter-fn (cond
                          (nil? (.wrapper src))
                          converter

                          (#{'seq 'vector} (.wrapper src))
                          (fn [_ d] (seq-converter d))

                          (= 'stream (.wrapper src))
                          (fn [_ d] (stream-converter d)))]

      ;; TODO: do a reverse traversal, not an exhaustive forward search
       (let [[src' dst'] (->> @src->dst->transfer
                              keys
                              (map (fn [src']
                                     (and
                                      (converter-fn src src')
                                      (when-let [dst' (some
                                                       #(and (converter-fn dst %) %)
                                                       (keys (@src->dst->transfer src')))]
                                        [src' dst']))))
                              (remove nil?)
                              first)]
         (cond
           (and src' dst')
           (let [f (get-in @src->dst->transfer [src' dst'])]
             (fn [source sink options]
               (let [source' (convert source src' options)
                     sink' (convert sink dst' options)]
                 (f source' sink' options))))

           (and (converter-fn src (g/type #'proto/ByteSource))
                (converter dst (g/type #'proto/ByteSink)))
           (fn [source sink {:keys [close?] :or {close? true} :as options}]
             (let [source' (convert source #'proto/ByteSource options)
                   sink' (convert sink #'proto/ByteSink options)]
               (default-transfer source' sink' options)
               (when close?
                 (doseq [x [source sink source' sink']]
                   (when (proto/closeable? x)
                     (proto/close x))))))

           :else
           nil))))))

;; for byte transfers
(defn transfer
  "Transfers, if possible, all bytes from `source` into `sink`.  If this cannot be accomplished, an IllegalArgumentException is
   thrown.

   `options` is a map whose available settings depends on the source and sink types:

   `chunk-size` - if a stream is being transformed into a sequence of discrete chunks, `:chunk-size` describes the
                  size of the chunks, which default to 4096 bytes.

   `encoding`   - if a string is being encoded or decoded, `:encoding` describes the charset that is used, which
                  defaults to 'UTF-8'

   `append?`    - if a file is being written to, `:append?` determines whether the bytes will overwrite the existing content
                  or be appended to the end of the file.  This defaults to true.

   `close?`     - whether the sink and source should be closed once the transfer is done, defaults to true."
  ([source sink]
   (transfer source sink nil))
  ([source sink options]
   (transfer source nil sink options))
  ([source source-type sink options]
   (let [src (type-descriptor source)
         dst (type-descriptor sink)]
     (if-let [f (transfer-fn src dst)]
       (f source sink options)
       (throw (IllegalArgumentException. (str "Don't know how to transfer between " (g/pprint-type src) " to " (g/pprint-type dst))))))))

(def ^{:doc "Web-scale."} dev-null
  (reify proto/ByteSink
    (send-bytes! [_ _ _])))

(defn optimized-transfer?
  "Returns true if an optimized transfer function exists for the given source and sink objects."
  [type-descriptor sink-type]
  (boolean (transfer-fn type-descriptor sink-type)))

;;; conversion definitions

(def-conversion ^{:cost 0} [(stream-of bytes) InputStream]
  [s options]
  (let [ps (ps/pushback-stream (get options :buffer-size 1024))]
    (d/loop []
      (d/chain (s/take! s ::none)
               (fn [^bytes msg]
                 (if (identical? ::none msg)
                   (do
                     (ps/close ps)
                     false)
                   (ps/put-array ps msg 0 (alength msg))))
               (fn [result]
                 (when result
                   (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))]
    (d/loop []
      (d/chain (s/take! s ::none)
               (fn [^ByteBuffer msg]
                 (if (identical? ::none msg)
                   (do
                     (ps/close ps)
                     false)
                   (ps/put-buffer ps (.duplicate msg))))
               (fn [result]
                 (when result
                   (d/recur)))))
    (ps/->input-stream ps)))

;; byte-array => byte-buffer
(def-conversion ^{:cost 0} [bytes ByteBuffer]
  [ary {:keys [direct?] :or {direct? false}}]
  (if direct?
    (let [len (Array/getLength ary)
          ^ByteBuffer buf (ByteBuffer/allocateDirect len)]
      (.put buf ary 0 len)
      (.position buf 0)
      buf)
    (ByteBuffer/wrap ary)))

;; byte-array => input-stream
(def-conversion ^{:cost 0} [bytes InputStream]
  [ary]
  (ByteArrayInputStream. ary))

;; byte-buffer => input-stream
(def-conversion ^{:cost 0} [ByteBuffer InputStream]
  [buf]
  (ByteBufferInputStream. (.duplicate buf)))

;; byte-buffer => byte-array
(def-conversion [ByteBuffer bytes]
  [buf]
  (if (.hasArray buf)
    (if (== (alength (.array buf)) (.remaining buf))
      (.array buf)
      (let [ary (Utils/byteArray (.remaining buf))]
        (doto buf
          .mark
          (.get ary 0 (.remaining buf))
          .reset)
        ary))
    (let [^bytes ary (Utils/byteArray (.remaining buf))]
      (doto buf .mark (.get ary) .reset)
      ary)))

;; sequence of byte-buffers => byte-buffer
(def-conversion [(vector-of ByteBuffer) ByteBuffer]
  [bufs {:keys [direct?] :or {direct? false}}]
  (cond
    (empty? bufs)
    (ByteBuffer/allocate 0)

    (and (empty? (rest bufs)) (not (proto/closeable? bufs)))
    (first bufs)

    :else
    (let [len (reduce + (map #(.remaining ^ByteBuffer %) bufs))
          buf (if direct?
                (ByteBuffer/allocateDirect len)
                (ByteBuffer/allocate len))]
      (doseq [^ByteBuffer b bufs]
        (.mark b)
        (.put buf b)
        (.reset b))
      (when (proto/closeable? bufs)
        (proto/close bufs))
      (.flip buf))))

;; byte-buffer => sequence of byte-buffers
(def-conversion ^{:cost 0} [ByteBuffer (vector-of ByteBuffer)]
  [buf {:keys [^long chunk-size]}]
  (if chunk-size
    (let [lim (.limit buf)
          indices (range (.position buf) lim chunk-size)]
      (mapv
       #(-> buf
            .duplicate
            ^ByteBuffer (.position (int %))
            ^ByteBuffer (.limit (int (min lim (p/+ (int %) chunk-size))))
            .slice)
       indices))
    [buf]))

;; channel => input-stream
(def-conversion ^{:cost 0} [ReadableByteChannel InputStream]
  [channel]
  (Channels/newInputStream channel))

;; 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}]
  (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]
  [input-stream]
  (Channels/newChannel input-stream))

;; string => byte-array
(def-conversion ^{:cost 2} [String byte-array-type]
  [s {:keys [encoding] :or {encoding "UTF-8"}}]
  (.getBytes s ^String (name encoding)))

;; byte-array => string
(def-conversion ^{:cost 2} [bytes String]
  [ary {:keys [encoding] :or {encoding "UTF-8"}}]
  (String. ^bytes ary (name encoding)))

;; lazy-seq of byte-buffers => channel
(def-conversion ^{:cost 1.5} [(seq-of ByteBuffer) ReadableByteChannel]
  [bufs]
  (let [pipe (Pipe/open)
        ^WritableByteChannel sink (.sink pipe)
        source (doto ^AbstractSelectableChannel (.source pipe)
                 (.configureBlocking true))]
    (future
      (try
        (loop [s bufs]
          (when (and (not (empty? s)) (.isOpen sink))
            (let [buf (.duplicate ^ByteBuffer (first s))]
              (.write sink buf)
              (recur (rest s)))))
        (finally
          (.close sink))))
    source))

(def-conversion ^{:cost 1.5} [(seq-of #'proto/ByteSource) InputStream]
  [srcs options]
  (let [chunk-size (get options :chunk-size 65536)
        out (PipedOutputStream.)
        in (PipedInputStream. out chunk-size)]
    (future
      (try
        (loop [s srcs]
          (when-not (empty? s)
            (transfer (first s) out)
            (recur (rest s))))
        (finally
          (.close out))))
    in))

(def-conversion ^{:cost 1.5} [InputStream byte-array-type]
  [in options]
  (let [out (ByteArrayOutputStream. (p/max 64 (.available in)))
        buf (Utils/byteArray 16384)]
    (loop []
      (let [len (.read in buf 0 16384)]
        (when-not (neg? len)
          (.write out buf 0 len)
          (recur))))
    (.toByteArray out)))

#_(let [ary (Utils/byteArray 0)]
    (def-conversion ^{:cost 0} [::nil byte-array-type]
      [src options]
      ary))

(def-conversion ^{:cost 2} [#'proto/ByteSource byte-array-type]
  [src options]
  (let [os (ByteArrayOutputStream.)]
    (transfer src os)
    (.toByteArray os)))

;; generic byte-source => lazy char-sequence
(def-conversion ^{:cost 2} [#'proto/ByteSource CharSequence]
  [source options]
  (cs/decode-byte-source
   #(when-let [bytes (proto/take-bytes! source % options)]
      (convert bytes ByteBuffer options))
   #(when (proto/closeable? source)
      (proto/close source))
   options))

;; input-stream => reader
(def-conversion ^{:cost 1.5} [InputStream Reader]
  [input-stream {:keys [encoding] :or {encoding "UTF-8"}}]
  (BufferedReader. (InputStreamReader. input-stream ^String encoding)))

;; reader => char-sequence
(def-conversion ^{:cost 1.5} [Reader CharSequence]
  [reader {:keys [chunk-size] :or {chunk-size 2048}}]
  (let [ary (char-array chunk-size)
        sb (StringBuilder.)]
    (loop []
      (let [n (.read reader ary 0 chunk-size)]
        (if (pos? n)
          (do
            (.append sb ary 0 n)
            (recur))
          (.toString sb))))))

;; char-sequence => string
(def-conversion [CharSequence String]
  [char-sequence]
  (.toString char-sequence))

(def-conversion [(vector-of String) String]
  [strings]
  (let [sb (StringBuilder.)]
    (doseq [s strings]
      (.append sb s))
    (.toString sb)))

;; file => readable-channel
(def-conversion ^{:cost 0} [File ReadableByteChannel]
  [file]
  (-> file
      (.toPath)
      (FileChannel/open (into-array StandardOpenOption
                                    [StandardOpenOption/READ]))))

;; file => writable-channel
(def-conversion ^{:cost 0} [File WritableByteChannel]
  [file {:keys [append?] :or {append? true}}]
  (let [option-array (into-array StandardOpenOption
                                 (cond-> [StandardOpenOption/CREATE
                                          StandardOpenOption/WRITE]
                                   append?
                                   (conj StandardOpenOption/APPEND)))]
    (-> file
        (.toPath)
        (FileChannel/open option-array))))

;;
(def-conversion ^{:cost 0
                  :doc "Assumes the file size is static."} [File (seq-of ByteBuffer)]
  [file {:keys [^long chunk-size writable?] :or {chunk-size (int 2e9), writable? false}}]
  (let [option-array (into-array StandardOpenOption
                                 (cond-> [StandardOpenOption/READ]
                                   writable?
                                   (conj StandardOpenOption/CREATE
                                         StandardOpenOption/WRITE)))
        ^FileChannel fc (-> file
                            (.toPath)
                            (FileChannel/open option-array))
        close-fn #(.close fc)
        buf-seq (fn buf-seq [^long offset]
                  (when (and (.isOpen fc)
                             (> (.size fc) offset))
                    (let [remaining (- (.size fc) offset)
                          close-after-reading? (<= remaining chunk-size)]
                      (lazy-seq
                       (cons
                        (let [mbb (.map fc
                                        (if writable?
                                          FileChannel$MapMode/READ_WRITE
                                          FileChannel$MapMode/READ_ONLY)
                                        offset
                                        (min remaining chunk-size))]
                          (when close-after-reading?
                            (close-fn))
                          mbb)
                        (buf-seq (+ offset chunk-size)))))))]
    (g/closeable-seq
     (buf-seq 0)
     false
     close-fn)))

;; output-stream => writable-channel
(def-conversion ^{:cost 0} [OutputStream WritableByteChannel]
  [output-stream]
  (Channels/newChannel output-stream))

;; writable-channel => output-stream
(def-conversion ^{:cost 0} [WritableByteChannel OutputStream]
  [channel]
  (Channels/newOutputStream channel))

;;; def-transfers

(def-transfer [ReadableByteChannel File]
  [channel file {:keys [chunk-size] :or {chunk-size (int 1e7)} :as options}]
  (let [^FileChannel fc (convert file WritableByteChannel options)]
    (try
      ;; .position() is 0 for normal mode, and the current size of the file in append mode
      (loop [pos (.position fc)]
        (let [n (.transferFrom fc channel pos chunk-size)]
          (when (pos? n)                                    ; 0 = end-of-channel
            (recur (+ pos n)))))
      (finally
        (.force fc true)
        (.close fc)))))

(def-transfer [File WritableByteChannel]
  [file
   channel
   {:keys [chunk-size
           close?]
    :or {chunk-size (int 1e6)
         close? true}
    :as options}]
  (let [^FileChannel fc (convert file ReadableByteChannel options)]
    (try
      (loop [idx 0]
        (let [n (.transferTo fc idx chunk-size channel)]
          (when (pos? n)
            (recur (+ idx n)))))
      (finally
        (when close?
          (.close ^WritableByteChannel channel))
        (.close fc)))))

(def-transfer [InputStream OutputStream]
  [input-stream
   output-stream
   {:keys [chunk-size
           close?]
    :or {chunk-size 4096
         close? true}
    :as options}]
  (let [ary (Utils/byteArray chunk-size)]
    (try
      (loop []
        (let [n (.read ^InputStream input-stream ary)]
          (when (pos? n)
            (.write ^OutputStream output-stream ary 0 n)
            (recur))))
      (.flush ^OutputStream output-stream)
      (finally
        (.close ^InputStream input-stream)
        (when close?
          (.close ^OutputStream output-stream))))))

;;; protocol extensions

(extend-protocol proto/ByteSink

  OutputStream
  (send-bytes! [this b _]
    (let [^OutputStream os this]
      (.write os ^bytes (convert b byte-array-type))))

  WritableByteChannel
  (send-bytes! [this b _]
    (let [^WritableByteChannel ch this]
      (.write ch ^ByteBuffer (convert b ByteBuffer)))))

(extend-protocol proto/ByteSource

  InputStream
  (take-bytes! [this n _]
    (let [ary (clojure.core/byte-array n)
          n (long n)]
      (loop [idx 0]
        (if (== idx n)
          ary
          (let [read (.read this ary idx (long (- n idx)))]
            (if (== -1 read)
              (when (pos? idx)
                (let [ary' (clojure.core/byte-array idx)]
                  (System/arraycopy ary 0 ary' 0 idx)
                  ary'))
              (recur (long (+ idx read)))))))))

  ReadableByteChannel
  (take-bytes! [this n {:keys [direct?] :or {direct? false}}]
    (let [^ByteBuffer buf (if direct?
                            (ByteBuffer/allocateDirect n)
                            (ByteBuffer/allocate n))]

      (loop []
        (when (try
                (pos? (.read this buf))
                (catch Throwable e
                  false))
          (recur)))

      (when (pos? (.position buf))
        (.flip buf))))

  ByteBuffer
  (take-bytes! [this ^long n _]
    (when (pos? (.remaining this))
      (let [n (int (min (.remaining this) n))
            buf (-> this
                    .duplicate
                    ^ByteBuffer (.limit (+ (.position this) n))
                    ^ByteBuffer (.slice)
                    (.order (.order this)))]
        (.position this (+ n (.position this)))
        buf))))

;;; print-bytes

(let [special-character? (->> "' _-+=`~{}[]()\\/#@!?.,;\"" (map int) set)]
  (defn- readable-character? [x]
    (or
     (Character/isLetterOrDigit (int x))
     (special-character? (int x)))))

(defn print-bytes
  "Prints out the bytes in both hex and ASCII representations, 16 bytes per line."
  [bytes]
  (let [bufs (convert bytes (seq-of ByteBuffer) {:chunk-size 16})]
    (doseq [^ByteBuffer buf bufs]
      (let [s (convert (.duplicate buf) String {:encoding "ISO-8859-1"})
            bytes (repeatedly (min 16 (.remaining buf)) #(.get buf))
            padding (* 3 (- 16 (count bytes)))
            hex-format #(->> "%02X" (repeat %) (interpose " ") (apply str))]
        (println
         (apply format
                (str
                 (hex-format (min 8 (count bytes)))
                 "  "
                 (hex-format (max 0 (- (count bytes) 8))))
                bytes)
         (apply str (repeat padding " "))
         "   "
         (->> s
              (map #(if (readable-character? %) % "."))
              (apply str)))))))

;;; to-* helpers

(defn ^ByteBuffer to-byte-buffer
  "Converts the object to a `java.nio.ByteBuffer`."
  ([x]
   (to-byte-buffer x nil))
  ([x options]
   (condp instance? x
     ByteBuffer x
     byte-array-type (ByteBuffer/wrap x)
     String (ByteBuffer/wrap (.getBytes ^String x (name (get options :encoding "UTF-8"))))
     (convert x ByteBuffer options))))

(defn to-byte-buffers
  "Converts the object to a sequence of `java.nio.ByteBuffer`."
  ([x]
   (to-byte-buffers x nil))
  ([x options]
   (convert x (seq-of ByteBuffer) options)))

(defn ^"[B" to-byte-array
  "Converts the object to a byte-array."
  ([x]
   (to-byte-array x nil))
  ([x options]
   (condp instance? x
     byte-array-type x
     String (.getBytes ^String x (name (get options :encoding "UTF-8")))
     (convert x byte-array-type options))))

(defn to-byte-arrays
  "Converts the object to a byte-array."
  ([x]
   (to-byte-array x nil))
  ([x options]
   (convert x (seq-of byte-array-type) options)))

(defn ^InputStream to-input-stream
  "Converts the object to a `java.io.InputStream`."
  ([x]
   (to-input-stream x nil))
  ([x options]
   (condp instance? x
     byte-array-type (ByteArrayInputStream. x)
     ByteBuffer (ByteBufferInputStream. x)
     (convert x InputStream options))))

(defn ^DataInputStream to-data-input-stream
  ([x]
   (to-data-input-stream x nil))
  ([x options]
   (if (instance? DataInputStream x)
     x
     (DataInputStream. (to-input-stream x)))))

(defn ^InputStream to-output-stream
  "Converts the object to a `java.io.OutputStream`."
  ([x]
   (to-output-stream x nil))
  ([x options]
   (convert x OutputStream options)))

(defn ^CharSequence to-char-sequence
  "Converts to the object to a `java.lang.CharSequence`."
  ([x]
   (to-char-sequence x nil))
  ([x options]
   (if (instance? CharSequence x)
     x
     (convert x CharSequence options))))

(defn ^ReadableByteChannel to-readable-channel
  "Converts the object to a `java.nio.ReadableByteChannel`"
  ([x]
   (to-readable-channel x nil))
  ([x options]
   (convert x ReadableByteChannel options)))

(defn ^String to-string
  "Converts the object to a string.
  Will consume all input it can.
  Consider using to-reader, to-line-seq, or to-char-sequence for lazy / partial input consumption."
  ([x]
   (to-string x nil))
  ([x options]
   (let [encoding (get options :encoding "UTF-8")]
     (condp instance? x
       String x
       byte-array-type (String. ^"[B" x ^String (name encoding))
       (convert x String options)))))

(defn to-reader
  "Converts the object to a java.io.Reader."
  ([x]
   (to-reader x nil))
  ([x options]
   (convert x Reader options)))

(defn to-line-seq
  "Converts the object to a lazy sequence of newline-delimited strings."
  ([x]
   (to-line-seq x nil))
  ([x options]
   (let [reader (convert x Reader options)
         reader (BufferedReader. ^Reader reader)
         line! (fn line! []
                 (lazy-seq
                  (when-let [l (try
                                 (.readLine reader)
                                 (catch IOException e
                                   nil))]
                    (cons l (line!)))))]
     (line!))))

(defn to-byte-source
  "Converts the object to something that satisfies `ByteSource`."
  ([x]
   (to-byte-source x nil))
  ([x options]
   (convert x #'proto/ByteSource options)))

(defn to-byte-sink
  "Converts the object to something that satisfies `ByteSink`."
  ([x]
   (to-byte-sink x nil))
  ([x options]
   (convert x #'proto/ByteSink options)))

;;;

(defn- cmp-bufs
  ^long [^ByteBuffer a' ^ByteBuffer b']
  (let [diff (p/- (.remaining a') (.remaining b'))
        sign (long (if (pos? diff) -1 1))
        a (if (pos? diff) b' a')
        b (if (pos? diff) a' b')
        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 (p/+ idx a-offset)))
                             (p/int->uint (.getInt b (p/+ idx b-offset))))]
                    (if (p/== 0 cmp)
                      (recur (p/+ idx 4))
                      ;; Use (if (pos? cmp) 1 -1) to ensure that the
                      ;; sign of the value x returned by cmp-bufs (and
                      ;; compare-bytes) is not modified when Clojure's
                      ;; comparator infrastructure calls (.intValue
                      ;; x).  The intValue method truncates a Java
                      ;; Long's most significant 32 bits away, which
                      ;; in some cases changes the sign of the result,
                      ;; and thus the direction of the comparison
                      ;; result.  Such code is not needed when
                      ;; comparing individual bytes below, because the
                      ;; subtraction result fits within the least
                      ;; significant 9 bits, and (.intValue x) never
                      ;; changes the sign.
                      (p/* sign (if (pos? cmp) 1 -1))))))]
      (if (p/== 0 (long cmp))
        (let [limit' (.remaining a)]
          (loop [idx limit]
            (if (p/>= idx limit')
              diff
              (let [cmp (p/-
                         (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))))))
        cmp))))

(defn compare-bytes
  "Returns a comparison result for two byte streams."
  ^long [a b]
  (if (and
       (or
        (instance? byte-array-type a)
        (instance? ByteBuffer a)
        (instance? String a))
       (or
        (instance? byte-array-type b)
        (instance? ByteBuffer b)
        (instance? String b)))
    (cmp-bufs (to-byte-buffer a) (to-byte-buffer b))
    (loop [a (to-byte-buffers a), b (to-byte-buffers b)]
      (cond
        (empty? a)
        (if (empty? b) 0 -1)

        (empty? b)
        1

        :else
        (let [cmp (cmp-bufs (first a) (first b))]
          (if (p/== 0 cmp)
            (recur (rest a) (rest b))
            cmp))))))

(defn bytes=
  "Returns true if the two byte streams are equivalent."
  [a b]
  (p/== 0 (compare-bytes a b)))



(comment
  (require '[manifold.stream :as ms])

  (def content
    (doto (ms/stream)
      (ms/put! (clojure.core/byte-array 5))
      (ms/close!)))

  (conversion-path (stream-of bytes) (seq-of bytes))        ; => ([(stream-of [B) (seq-of [B)])

  (convert content (seq-of bytes))  ; doesn't work
  (= content (convert content (seq-of bytes)))

  (convert content (seq-of bytes) {:source-type (stream-of bytes)})  ; works
  (= content (convert content (seq-of bytes) {:source-type (stream-of bytes)}))

  (convert content (seq-of bytes) {:source-type (stream-of nil)})  ; doesn't work
  (= content (convert content (seq-of bytes) {:source-type (stream-of nil)}))

  (convert content (seq-of String)) ; also doesn't work
  (= content (convert content (seq-of String)))

  (convert content (seq-of java.nio.ByteBuffer)) ; this works
  (= content (convert content (seq-of java.nio.ByteBuffer)))

  (let [content (doto (ms/stream)
                  (ms/put! (clojure.core/byte-array 5))
                  (ms/close!))]
    (convert content (seq-of bytes))  ; no
    #_(convert content (seq-of bytes) {:source-type (stream-of bytes)}) ; yes
    #_(convert content (seq-of bytes) {:source-type (stream-of nil)}) ; no
    )
  (let [content (doto (ms/stream)
                  (ms/put! (clojure.core/byte-array 5))
                  (ms/close!))]
    (convert content (seq-of String)) ; also doesn't work
    (= content (convert content (seq-of String))))

  (let [content (doto (ms/stream)
                  (ms/put! (clojure.core/byte-array 5))
                  (ms/close!))]
    (convert content (seq-of java.nio.ByteBuffer)) ; this works
    (= content (convert content (seq-of java.nio.ByteBuffer))))

  )