fix transducer composition problems, add tests

remove `preserving-reduced` from several transducers. This was modifying
the reducing function passed to the transducer, causing reduced values
to be wrapped in `reduced` twice, so the reducing process wasn't
completely unwrapping reduced values when
finishing. Reducing functions only need to be wrapped with
`preserving-reduced` if they call `reduce` themselves. (Like `cat` and
`pixie.io.common/stream-reducer`)

Also fixes a problem with the `take` transducer consuming one more
element than needed.

Add tests for these problems.

Author: notnew

pixie/io.pxi

@@ -66,14 +66,13 @@
             (str string))))))
   IReduce
   (-reduce [this f init]
-    (let [rrf (preserving-reduced f)]
-      (loop [acc init]
-        (if-let [line (-read-line this)]
-          (let [result (rrf acc line)]
-            (if (not (reduced? result))
-              (recur result)
-              @result))
-          acc)))))
+    (loop [acc init]
+      (if-let [line (-read-line this)]
+        (let [result (f acc line)]
+          (if (reduced? result)
+            @result
+            (recur result)))
+        acc))))
 
 (defn line-reader
   [input-stream]
@@ -150,14 +149,13 @@
 (deftype BufferedInputStream [upstream idx buffer]
   IReduce
   (-reduce [this f init]
-    (let [rrf (preserving-reduced f)]
-      (loop [acc init]
-        (if-let [next-byte (read-byte this)]
-          (let [step (rrf acc next-byte)]
-            (if (reduced? step)
-              @step
-              (recur step)))
-          acc))))
+    (loop [acc init]
+      (if-let [next-byte (read-byte this)]
+        (let [result (f acc next-byte)]
+          (if (reduced? result)
+            @result
+            (recur result)))
+        acc)))
   IByteInputStream
   (read-byte [this]
     (when (= idx (count buffer))

pixie/stdlib.pxi

@@ -1640,21 +1640,25 @@ The new value is thus `(apply f current-value-of-atom args)`."
        (let [idx (if (neg? i) (+ i (count coll)) i)]
          (nth coll idx not-found)))))
 
+(defn ensure-reduced [x]
+  (if (reduced? x)
+    x
+    (reduced x)))
+
 (defn take
   {:doc "Takes n elements from the collection, or fewer, if not enough."
    :added "0.1"}
   ([n]
    (fn [rf]
-     (let [rrf (preserving-reduced rf)
-           seen (atom 0)]
+     (let [seen (atom 0)]
        (fn
          ([] (rf))
          ([result] (rf result))
          ([result input]
           (let [s (swap! seen inc)]
-            (if (<= s n)
-              (rrf result input)
-              (reduced result))))))))
+            (cond (< s n) (rf result input)
+                  (= s n) (ensure-reduced (rf result input))
+                  :else (reduced result))))))))
   ([n coll]
    (lazy-seq
      (when (pos? n)
@@ -1666,16 +1670,15 @@ The new value is thus `(apply f current-value-of-atom args)`."
    :added "0.1"}
   ([n]
    (fn [rf]
-     (let [rrf (preserving-reduced rf)
-           seen (atom 0)]
+     (let [seen (atom 0)]
        (fn
          ([] (rf))
          ([result]
           (rf result))
          ([result input]
           (let [s (swap! seen inc)]
             (if (> s n)
-              (rrf result input)
+              (rf result input)
               result)))))))
   ([n coll]
    (let [s (seq coll)]
@@ -1707,14 +1710,13 @@ The new value is thus `(apply f current-value-of-atom args)`."
   :added "0.1"}
   ([pred]
      (fn [rf]
-       (let [rrf (preserving-reduced rf)]
-         (fn
-           ([] (rf))
-           ([result] (rf result))
-           ([result input]
+       (fn
+         ([] (rf))
+         ([result] (rf result))
+         ([result input]
             (if (pred input)
-              (rrf result input)
-              (reduced result)))))))
+              (rf result input)
+              (reduced result))))))
   ([pred coll]
      (lazy-seq
       (when-let [s (seq coll)]
@@ -1825,13 +1827,12 @@ not enough elements were present."
    :signatures [[f] [f coll]]}
   ([f]
    (fn [rf]
-     (let [i (atom -1)
-           rrf (preserving-reduced rf)]
+     (let [i (atom -1)]
        (fn
          ([] (rf))
          ([result] (rf result))
          ([result input]
-          (rrf result (f (swap! i inc) input)))))))
+          (rf result (f (swap! i inc) input)))))))
   ([f coll]
    (let [mapi (fn mapi [i coll]
                 (lazy-seq
@@ -1849,8 +1850,7 @@ not enough elements were present."
    :added "0.1"}
   ([f]
    (fn [rf]
-     (let [iv (atom -1)
-           rrf (preserving-reduced rf)]
+     (let [iv (atom -1)]
        (fn
          ([] (rf))
          ([result] (rf result))
@@ -1859,7 +1859,7 @@ not enough elements were present."
                 v (f i input)]
             (if (nil? v)
               result
-              (rrf result v))))))))
+              (rf result v))))))))
   ([f coll]
    (let [keepi (fn keepi [i coll]
                  (lazy-seq

tests/pixie/tests/test-io.pxi

@@ -3,18 +3,33 @@
   (require pixie.streams :as st :refer :all)
   (require pixie.streams.utf8 :as utf8 :refer :all)
   (require pixie.io :as io)
+  (require pixie.io-blocking :as blocking)
   (require pixie.streams.zlib :as zlib))
 
 (t/deftest test-file-reduction
   (let [f (io/open-read "tests/pixie/tests/test-io.txt")]
     (t/assert= (transduce (map identity)
                           count-rf
                           f)
-               91)))
+               91))
+  (let [f (io/open-read "tests/pixie/tests/test-io.txt")]
+    (t/assert= (transduce (comp (map char) (take 4)) string-builder f)
+               "This"))
+  (let [f (blocking/open-read "tests/pixie/tests/test-io.txt")]
+    (t/assert= (transduce (map identity)
+                          count-rf
+                          f)
+               91))
+  (let [f (blocking/open-read "tests/pixie/tests/test-io.txt")]
+    (t/assert= (transduce (comp (map char) (take 4)) string-builder f)
+               "This")))
 
 (t/deftest test-process-reduction
   (let [f (io/run-command "ls tests/pixie/tests/test-io.txt")]
-    (t/assert= f "tests/pixie/tests/test-io.txt\n")))
+    (t/assert= f "tests/pixie/tests/test-io.txt\n"))
+  (let [pipe (blocking/popen-read "ls tests/pixie/tests/test-io.txt")]
+    (t/assert= (transduce (comp (map char) (take 6)) string-builder pipe)
+               "tests/")))
 
 (t/deftest test-read-into-buffer
   (let [f (io/open-read "tests/pixie/tests/test-io.txt")]
@@ -38,6 +53,14 @@
     (t/assert= (io/read-line f) "Second line.")
     (t/assert= (io/read-line f) nil)))
 
+(t/deftest test-line-reader
+  (let [lines (io/line-reader (io/open-read "tests/pixie/tests/test-io.txt"))]
+    (t/assert= (transduce (map count) conj [] lines)
+               [77 12]))
+  (let [lines (io/line-reader (io/open-read "tests/pixie/tests/test-io.txt"))]
+    (t/assert= (transduce (comp (map count) (take 1)) conj [] lines)
+               [77])))
+
 (t/deftest test-line-seq
   (let [f (io/buffered-input-stream (io/open-read "tests/pixie/tests/test-io.txt"))
         s (io/line-seq f)]
@@ -58,7 +81,11 @@
     (t/assert= (char (io/read-byte f)) \T)
     (t/assert= (char (io/read-byte f)) \h)
     (t/assert= (char (io/read-byte f)) \i)
-    (t/assert= (char (io/read-byte f)) \s)))
+    (t/assert= (char (io/read-byte f)) \s)
+    (t/assert= (transduce (comp (map char) (take 5)) string-builder f)
+               " is a")
+    (t/assert= (transduce (comp (map char) (take 5)) string-builder f)
+               " test")))
 
 (t/deftest test-buffered-input-streams-throws-on-non-input-streams
   (let [f (io/buffered-input-stream (io/open-read "tests/pixie/tests/test-io.txt"))]

tests/pixie/tests/test-stdlib.pxi

@@ -28,9 +28,28 @@
   (t/assert= (keep-indexed (constantly nil) []) [])
   (t/assert= (keep-indexed (fn [i x] [i x]) [:a :b]) [[0 :a] [1 :b]])
 
-  (t/assert= (transduce (keep-indexed (constantly true)) conj []) [])
-  (t/assert= (transduce (keep-indexed (constantly nil)) conj []) [])
-  (t/assert= (transduce (keep-indexed (fn [i x] [i x])) conj [:a :b]) [[0 :a] [1 :b]]))
+  (t/assert= (transduce (keep-indexed (constantly true)) conj [:a :b]) [true true])
+  (t/assert= (transduce (keep-indexed (constantly nil)) conj [:a :b]) [])
+  (t/assert= (transduce (keep-indexed (fn [i x] [i x])) conj [:a :b]) [[0 :a] [1 :b]])
+
+  (let [even-index? (fn [i x]
+                        (when (even? i)
+                          x))]
+    (t/assert= (transduce (keep-indexed even-index?) conj [:a :b :c :d])
+               [:a :c])
+    (t/assert= (transduce (map-indexed even-index?) conj [:a :b :c :d])
+               [:a nil :c nil])
+
+    (t/assert= (transduce (comp (keep-indexed even-index?)
+                                (take 3))
+                          conj
+                          [:a :b :c :d :e :f])
+               [:a :c :e])
+    (t/assert= (transduce (comp (map-indexed even-index?)
+                                (take 3))
+                          conj
+                          [:a :b :c :d])
+               [:a nil :c])))
 
 (t/deftest test-reductions
   (t/assert= (reductions + nil)
@@ -542,7 +561,21 @@
   (t/assert= (transduce (take 0) conj [1 2 3 4]) [])
   (t/assert= (transduce (take 1) conj [1 2 3 4]) [1])
   (t/assert= (transduce (take 2) conj [1 2 3 4]) [1 2])
-  (t/assert= (transduce (take 3) conj [1 2 3 4]) [1 2 3]))
+  (t/assert= (transduce (take 3) conj [1 2 3 4]) [1 2 3])
+  (t/assert= (transduce (take 10) conj [1 2 3 4]) [1 2 3 4])
+  (t/assert= (transduce (comp (take 2) (take 1)) conj [1 2 3 4]) [1])
+  (t/assert= (transduce (comp (take 1) (take 2)) conj [1 2 3 4]) [1])
+
+  (let [call-count (atom 0)
+        inc-call-count! (fn [x]
+                          (swap! call-count inc)
+                          x)]
+    (t/assert= (transduce (comp (map inc-call-count!) (take 2)) conj (range 10))
+               [0 1])
+    (t/assert= @call-count 2)
+    (t/assert= (transduce (comp (take 2) (map inc-call-count!)) conj (range 10))
+               [0 1])
+    (t/assert= @call-count 4)))
 
 (t/deftest test-drop
   (t/assert= (drop 0 [1 2 3 4]) [1 2 3 4])
@@ -552,21 +585,26 @@
   (t/assert= (transduce (drop 0) conj [1 2 3 4]) [1 2 3 4])
   (t/assert= (transduce (drop 1) conj [1 2 3 4]) [2 3 4])
   (t/assert= (transduce (drop 2) conj [1 2 3 4]) [3 4])
-  (t/assert= (transduce (drop 3) conj [1 2 3 4]) [4]))
+  (t/assert= (transduce (drop 3) conj [1 2 3 4]) [4])
+  (t/assert= (transduce (drop 10) conj [1 2 3 4]) [])
+  (t/assert= (transduce (comp (drop 1) (take 2)) conj [1 2 3 4]) [2 3])
+  (t/assert= (transduce (comp (take 2) (drop 1)) conj [1 2 3 4]) [2]))
 
 (t/deftest test-take-while
   (t/assert= (take-while pos? [1 2 3 -1]) [1 2 3])
   (t/assert= (take-while pos? [-1 2]) ())
   (t/assert= (transduce (take-while even?) conj [2 4 6 7 8]) [2 4 6])
   (t/assert= (transduce (take-while even?) conj [0 2] [1 4 6]) [0 2])
-  (t/assert= (transduce (take-while even?) conj [1 3] [2 4 6 7 8]) [1 3 2 4 6]))
+  (t/assert= (transduce (take-while even?) conj [1 3] [2 4 6 7 8]) [1 3 2 4 6])
+  (t/assert= (transduce (comp (take-while even?) (take 2)) conj [1 3] [2 4 6 7 8]) [1 3 2 4]))
 
 (t/deftest test-drop-while
   (t/assert= (drop-while pos? [1 2 3 -1]) [-1])
   (t/assert= (drop-while pos? [-1 2]) [-1 2])
   (t/assert= (transduce (drop-while even?) conj [2 4 6 7 8]) [7 8])
   (t/assert= (transduce (drop-while even?) conj [0 2] [1 4 6]) [0 2 1 4 6])
-  (t/assert= (transduce (drop-while even?) conj [0 2] [2 4 6 7 8]) [0 2 7 8]))
+  (t/assert= (transduce (drop-while even?) conj [0 2] [2 4 6 7 8]) [0 2 7 8])
+  (t/assert= (transduce (comp (drop-while even?) (take 2)) conj [0 2] [2 4 6 7 8]) [0 2 7 8]))
 
 (t/deftest test-cycle
   (t/assert= (cycle ()) ())