Bugfix/closeable seq finalize

.clj-kondo/config.edn

@@ -1,4 +1,8 @@
-{:lint-as {byte-streams.utils/defprotocol+ clojure.core/defprotocol
- byte-streams.utils/deftype+ clojure.core/deftype
- byte-streams.utils/defrecord+ clojure.core/defrecord
- byte-streams.utils/definterface+ clojure.core/definterface}}
+{:lint-as {byte-streams.utils/defprotocol+ clojure.core/defprotocol
+ byte-streams.utils/deftype+ clojure.core/deftype
+ byte-streams.utils/defrecord+ clojure.core/defrecord
+ byte-streams.utils/definterface+ clojure.core/definterface
+ clj-commons.byte-streams.utils/defprotocol+ clojure.core/defprotocol
+ clj-commons.byte-streams.utils/deftype+ clojure.core/deftype
+ clj-commons.byte-streams.utils/defrecord+ clojure.core/defrecord
+ clj-commons.byte-streams.utils/definterface+ clojure.core/definterface}}

.gitignore

@@ -16,3 +16,4 @@ dump*
 DS_Store
 .nrepl*
 .clj-kondo/.cache
+.eastwood

src/clj_commons/byte_streams.clj

@@ -9,46 +9,40 @@
  [protocols :as proto]
  [pushback-stream :as ps]
  [char-sequence :as cs]]
- [clojure.java.io :as io]
  [clj-commons.primitive-math :as p])
  (:import
- [clj_commons.byte_streams
- Utils
- ByteBufferInputStream]
- [clj_commons.byte_streams.graph
- Type]
- [java.nio
- ByteBuffer
- DirectByteBuffer]
- [java.lang.reflect
- Array]
- [java.util.concurrent.atomic
- AtomicBoolean]
- [java.io
- File
- FileOutputStream
- FileInputStream
- ByteArrayInputStream
- ByteArrayOutputStream
- PipedOutputStream
- PipedInputStream
- DataInputStream
- InputStream
- OutputStream
- IOException
- RandomAccessFile
- Reader
- InputStreamReader
- BufferedReader]
- [java.nio.channels
- ReadableByteChannel
- WritableByteChannel
- FileChannel
- FileChannel$MapMode
- Channels
- Pipe]
- [java.nio.channels.spi
- AbstractSelectableChannel]))
+ (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)))
 
 ;;;
 
@@ -576,35 +570,57 @@
 ;; file => readable-channel
 (def-conversion ^{:cost 0} [File ReadableByteChannel]
  [file]
- (.getChannel (FileInputStream. 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}}]
- (.getChannel (FileOutputStream. file (boolean append?))))
-
-(def-conversion ^{:cost 0} [File (seq-of ByteBuffer)]
+ (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 [chunk-size writable?] :or {chunk-size (int 2e9), writable? false}}]
- (let [^RandomAccessFile raf (RandomAccessFile. file (if writable? "rw" "r"))
- ^FileChannel fc (.getChannel raf)
+ (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 [offset]
- (when-not (<= (.size fc) offset)
- (let [remaining (- (.size fc) offset)]
+ (when (and (.isOpen fc)
+ (> (.size fc) offset))
+ (let [remaining (- (.size fc) offset)
+ close-after-reading? (<= remaining chunk-size)]
  (lazy-seq
  (cons
- (.map fc
- (if writable?
- FileChannel$MapMode/READ_WRITE
- FileChannel$MapMode/READ_ONLY)
- offset
- (min remaining chunk-size))
+ (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
- #(do
- (.close raf)
- (.close fc)))))
+ close-fn)))
 
 ;; output-stream => writable-channel
 (def-conversion ^{:cost 0} [OutputStream WritableByteChannel]

src/clj_commons/byte_streams/graph.clj

@@ -194,25 +194,21 @@
  (close-fn)
  nil)
  (reify
-
  clojure.lang.IPending
  (isRealized [_]
  (or
  (not (instance? clojure.lang.IPending s))
  (realized? s)))
 
- Object
- (finalize [_]
- (close-fn))
-
  java.io.Closeable
  (close [_]
  (close-fn))
 
  clojure.lang.Sequential
- clojure.lang.ISeq
  clojure.lang.Seqable
  (seq [this] this)
+
+ clojure.lang.ISeq
  (cons [_ a]
  (closeable-seq (cons a s) exhaustible? close-fn))
  (next [this]

test/clj_commons/byte_streams_test.clj

@@ -1,20 +1,16 @@
 (ns clj-commons.byte-streams-test
  (:require
- [clj-commons.byte-streams :refer [bytes= compare-bytes conversion-path convert dev-null possible-conversions seq-of stream-of to-byte-array to-byte-buffer to-byte-buffers to-input-stream to-string transfer vector-of]]
+ [clj-commons.byte-streams :refer [bytes= compare-bytes conversion-path convert dev-null possible-conversions seq-of stream-of to-byte-array to-byte-buffer to-byte-buffers to-input-stream to-string transfer vector-of] :as bs]
  [clojure.test :refer :all]
- [clj-commons.byte-streams.char-sequence :as cs])
+ [clojure.java.io :as io]
+ [clj-commons.primitive-math :as p])
  (:refer-clojure
  :exclude [vector-of])
  (:import
- [java.nio.charset
- Charset]
- [java.io
- ByteArrayInputStream
- File]
- [java.nio
- ByteBuffer]
- [java.util
- Arrays]))
+ (java.io File Closeable Reader)
+ (java.nio ByteBuffer)
+ (java.nio.channels WritableByteChannel ReadableByteChannel)
+ (java.util Arrays)))
 
 (def ^String text
  "The suburb of Saffron Park lay on the sunset side of London, as red and ragged as a cloud of sunset. It was built of a bright brick throughout; its sky-line was fantastic, and even its ground plan was wild. It had been the outburst of a speculative builder, faintly tinged with art, who called its architecture sometimes Elizabethan and sometimes Queen Anne, apparently under the impression that the two sovereigns were identical. It was described with some justice as an artistic colony, though it never in any definable way produced any art. But although its pretensions to be an intellectual centre were a little vague, its pretensions to be a pleasant place were quite indisputable. The stranger who looked for the first time at the quaint red houses could only think how very oddly shaped the people must be who could fit in to them. Nor when he met the people was he disappointed in this respect. The place was not only pleasant, but perfect, if once he could regard it not as a deception but rather as a dream. Even if the people were not \"artists,\" the whole was nevertheless artistic. That young man with the long, auburn hair and the impudent face—that young man was not really a poet; but surely he was a poem. That old gentleman with the wild, white beard and the wild, white hat—that venerable humbug was not really a philosopher; but at least he was the cause of philosophy in others. That scientific gentleman with the bald, egg-like head and the bare, bird-like neck had no real right to the airs of science that he assumed. He had not discovered anything new in biology; but what biological creature could he have discovered more singular than himself? Thus, and thus only, the whole place had properly to be regarded; it had to be considered not so much as a workshop for artists, but as a frail but finished work of art. A man who stepped into its social atmosphere felt as if he had stepped into a written comedy.")
@@ -50,7 +46,7 @@
  (str (pr-str src) " -> " (pr-str dst)))))
 
  ;; make sure none of our intermediate representations are strings if our target isn't a string
- (let [invalid-destinations (->> #{String CharSequence java.io.Reader}
+ (let [invalid-destinations (->> #{String CharSequence Reader}
  (mapcat #(vector % (list 'seq-of %) (list 'stream-of %)))
  set)
  pairwise-conversions (->> (class ary)
@@ -139,7 +135,95 @@
  (is (bytes= text-bytes (-> text-bytes (to-input-stream {:chunk-size 128}) to-string to-byte-array)))))
 
 (deftest compare-bytes-former-bug
- (let [bx (convert (byte-array [0x00 0x00 0x00 0x01]) java.nio.ByteBuffer)
- by (convert (byte-array [0x80 0x00 0x00 0x01]) java.nio.ByteBuffer)]
+ (let [bx (convert (byte-array [0x00 0x00 0x00 0x01]) ByteBuffer)
+ by (convert (byte-array [0x80 0x00 0x00 0x01]) ByteBuffer)]
  (is (= [bx by] (sort compare-bytes [bx by])))
  (is (= [bx by] (sort compare-bytes [by bx])))))
+
+(defn- write-zeros-file
+ "Write out a file of nothing but zeros"
+ [size]
+ (let [f (doto (File/createTempFile "byte-streams-test-" nil)
+ (.deleteOnExit))
+ buf-size 64
+ zs (byte-array buf-size (byte 0))
+ num-bufs (int (quot size buf-size))
+ remainder (int (mod size buf-size))]
+ (with-open [os (io/output-stream f)]
+ (loop [cnt num-bufs]
+ (when (p/> cnt 0)
+ (.write os zs)
+ (recur (p/dec cnt))))
+ (when (pos? remainder)
+ (.write os (byte-array remainder (byte 0)))))
+ f))
+
+(defn- bb-stream-size
+ [s]
+ (reduce
+ (fn [sz ^ByteBuffer bb] (p/+ ^int sz (.limit bb)))
+ (int 0)
+ s))
+
+(deftest large-file
+ (testing "can read whole file"
+ (let [size (int 1e8)
+ chunk-size (p// size 10)
+ f (write-zeros-file size)]
+ (testing "from streams"
+ (testing "all at once"
+ (with-open [in (io/input-stream f)]
+ (is (== (bb-stream-size (convert in (seq-of ByteBuffer)))
+ size))))
+ (testing "in chunks"
+ (with-open [in (io/input-stream f)]
+ (is (== (bb-stream-size (convert in
+ (seq-of ByteBuffer)
+ {:chunk-size chunk-size}))
+ size)))))
+
+ (testing "from java.io.File"
+ (is (== (bb-stream-size (convert f (seq-of ByteBuffer)))
+ size))
+
+ (testing "partial read, then close"
+ (let [n 4
+ s (convert f (seq-of ByteBuffer) {:chunk-size chunk-size})
+ _ (nth s n)]
+ (.close ^Closeable s)
+ (is (some? (nth s n)))
+ (is (thrown? Exception
+ (nth s (inc n))))))))))
+
+(deftest from-io-file
+ (testing "File conversions not tested elsewhere"
+ (testing "read/write to same spot"
+ (let [size 1
+ val (byte (rand-int 127))
+ val-bb (ByteBuffer/wrap (byte-array 1 val))
+ f (write-zeros-file size)]
+ (with-open [^WritableByteChannel wbc (convert f WritableByteChannel {:append? false})]
+ (.write wbc val-bb))
+ (with-open [^ReadableByteChannel rbc (convert f ReadableByteChannel)]
+ (let [bb (ByteBuffer/allocate 1)]
+ (.read rbc bb)
+ (is (bytes= val-bb bb))))))
+
+ (testing "append"
+ (let [^int size (rand-int 100)
+ val (byte (rand-int 127))
+ val-bb (ByteBuffer/wrap (byte-array 1 val))
+ f (write-zeros-file size)]
+ (with-open [^WritableByteChannel wbc (convert f WritableByteChannel {:append? true})]
+ (.write wbc val-bb))
+ (with-open [^ReadableByteChannel rbc (convert f ReadableByteChannel)]
+ (let [bb (ByteBuffer/allocate (inc size))]
+ (.read rbc bb)
+
+ (let [bb-array (.array bb)]
+ (dotimes [i size]
+ (is (= (aget bb-array i) 0)))
+ (is (= (aget bb-array size) val)))))))))
+
+
+