Merge pull request #256 from thomasmulvaney/seekable-streams

add ISeekableStream protocol

Author: halgari

pixie/io-blocking.pxi

@@ -3,6 +3,9 @@
 
 
 (def fopen (ffi-fn libc "fopen" [CCharP CCharP] CVoidP))
+(def fseek (ffi-fn libc "fseek" [CVoidP CInt CInt] CInt))
+(def ftell  (ffi-fn libc "ftell" [CVoidP] CInt))
+(def -rewind (ffi-fn libc "rewind" [CVoidP] CVoidP))
 (def fread (ffi-fn libc "fread" [CVoidP CInt CInt CVoidP] CInt))
 (def fgetc (ffi-fn libc "fgetc" [CVoidP] CInt))
 (def fputc (ffi-fn libc "fputc" [CInt CVoidP] CInt))
@@ -15,6 +18,18 @@
 
 (def DEFAULT-BUFFER-SIZE 1024)
 
+(defn default-stream-reducer [this f init]
+  (let [buf (buffer DEFAULT-BUFFER-SIZE)
+        rrf (preserving-reduced f)]
+    (loop [acc init]
+      (let [read-count (read this buf DEFAULT-BUFFER-SIZE)]
+        (if (> read-count 0)
+          (let [result (reduce rrf acc buf)]
+            (if (not (reduced? result))
+              (recur result)
+              @result))
+          acc)))))
+
 (deftype FileStream [fp]
   IInputStream
   (read [this buffer len]
@@ -25,21 +40,17 @@
       read-count))
   (read-byte [this]
     (fgetc buffer))
+  ISeekableStream
+  (seek [this pos]
+    (fseek fp pos 0))
+  (rewind [this]
+    (-rewind fp))
   IDisposable
   (-dispose! [this]
     (fclose fp))
   IReduce
   (-reduce [this f init]
-    (let [buf (buffer DEFAULT-BUFFER-SIZE)
-          rrf (preserving-reduced f)]
-      (loop [acc init]
-        (let [read-count (read this buf DEFAULT-BUFFER-SIZE)]
-          (if (> read-count 0)
-            (let [result (reduce rrf acc buf)]
-              (if (not (reduced? result))
-                (recur result)
-                @result))
-            acc))))))
+    (default-stream-reducer this f init)))
 
 (defn open-read
   {:doc "Open a file for reading, returning a IInputStream"
@@ -109,6 +120,15 @@
     (dispose! c)
     result))
 
+(defn slurp-stream [stream]
+  (let [c stream
+        result (transduce
+                (map char)
+                string-builder
+                c)]
+    (dispose! c)
+    result))
+
 (deftype ProcessInputStream [fp]
   IInputStream
   (read [this buffer len]
@@ -124,16 +144,7 @@
     (pclose fp))
   IReduce
   (-reduce [this f init]
-    (let [buf (buffer DEFAULT-BUFFER-SIZE)
-          rrf (preserving-reduced f)]
-      (loop [acc init]
-        (let [read-count (read this buf DEFAULT-BUFFER-SIZE)]
-          (if (> read-count 0)
-            (let [result (reduce rrf acc buf)]
-              (if (not (reduced? result))
-                (recur result)
-                @result))
-            acc))))))
+    (default-stream-reducer this f init)))
 
 
 (defn popen-read

pixie/io.pxi

@@ -44,6 +44,13 @@
       (set-field! this :offset (+ offset read-count))
       (set-buffer-count! buffer read-count)
       read-count))
+  ISeekableStream
+  (position [this]
+    offset)
+  (rewind [this]
+    (set-field! this :offset 0))
+  (seek [this pos]
+    (set-field! this :offset pos))
   IDisposable
   (-dispose! [this]
     (dispose! uvbuf)

pixie/streams.pxi

@@ -11,3 +11,8 @@
 
 (defprotocol IByteOutputStream
   (write-byte [this byte]))
+
+(defprotocol ISeekableStream
+  (position [this])
+  (seek [this position])
+  (rewind [this]))

tests/pixie/tests/test-io.pxi

@@ -1,5 +1,6 @@
 (ns pixie.tests.test-io
   (require pixie.test :as t)
+  (require pixie.streams :as st :refer :all)
   (require pixie.io :as io))
 
 (t/deftest test-file-reduction
@@ -24,6 +25,16 @@
         s (io/line-seq f)]
     (t/assert= (last s) "Second line.")))
 
+(t/deftest test-seek
+  (let [f (io/open-read "tests/pixie/tests/test-io.txt")]
+    (io/read-line f)
+    (t/assert= (io/read-line f) "Second line.")
+    (io/rewind f)
+    (io/read-line f)
+    (t/assert= (io/read-line f) "Second line.")
+    (io/seek f (- (position f) 6))
+    (t/assert= (io/read-line f) "line.")))
+
 (t/deftest test-slurp-spit
   (let [val (vec (range 1280))]
     (io/spit "test.tmp" val)