Manifold body stream to input-stream

manifold_input_stream.clj

(require '[byte-streams :as bs]
         '[manifold.deferred :as d]
         '[manifold.stream :as stream])
(import '[java.io ByteArrayInputStream])

(defn raw-stream->input-stream [byte-stream total-bytes]
  (let [bs-array (byte-array total-bytes)]
    (d/chain byte-stream
             #(stream/map bs/to-byte-array %)
             #(stream/reduce
               (fn [offset bytes]
                 (let [length (alength bytes)
                       new-length (+ offset length)]
                   (when (> new-length total-bytes)
                     (throw (ex-info "Reached max bytes" {:total-bytes total-bytes})))
                   (System/arraycopy bytes 0 bs-array offset length)
                   new-length))
               0
               %)
             (fn [length]
               (ByteArrayInputStream. bs-array 0 length)))))

(defn input-stream->string [input-stream]
   (with-open [in input-stream]
     (slurp in)))