CLJ-2775: Added support for java.util.stream streams including stream-reduce, stream-seq, stream-transduce, and stream-into.

Author: puredanger

src/clj/clojure/core.clj

@@ -6830,8 +6830,6 @@ fails, attempts to require sym's namespace and retries."
             `(let [~ge ~e] (case* ~ge ~shift ~mask ~default ~imap ~switch-type :hash-identity ~skip-check))))))))
 
 
-;; redefine reduce with internal-reduce
-
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; helper files ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 (alter-meta! (find-ns 'clojure.core) assoc :doc "Fundamental library of the Clojure language")
 (load "core_proxy")
@@ -6841,6 +6839,45 @@ fails, attempts to require sym's namespace and retries."
 (load "core/protocols")
 (load "gvec")
 
+(defn stream-reduce!
+  "Works like reduce but takes a java.util.stream.BaseStream as its source.
+  Honors 'reduced', is a terminal operation on the stream"
+  {:added "1.12"}
+  ([f ^java.util.stream.BaseStream s]
+   (clojure.core.protocols/iterator-reduce! (.iterator s) f))
+  ([f init ^java.util.stream.BaseStream s]
+   (clojure.core.protocols/iterator-reduce! (.iterator s) f init)))
+
+(defn stream-seq!
+  "Takes a java.util.stream.BaseStream instance s and returns a seq of its
+  contents. This is a terminal operation on the stream."
+  {:added "1.12"}
+  [^java.util.stream.BaseStream stream]
+  (iterator-seq (.iterator stream)))
+
+(defn stream-transduce!
+  "Works like transduce but takes a java.util.stream.BaseStream as its source.
+  This is a terminal operation on the stream."
+  {:added "1.12"}
+  ([xform f ^java.util.stream.BaseStream stream] (stream-transduce! xform f (f) stream))
+  ([xform f init ^java.util.stream.BaseStream stream]
+   (let [f (xform f)
+         ret (stream-reduce! f init stream)]
+     (f ret))))
+
+(defn stream-into!
+  "Returns a new coll consisting of coll with all of the items of the
+  stream conjoined. This is a terminal operation on the stream."
+  {:added "1.12"}
+  ([to ^java.util.stream.BaseStream stream]
+   (if (instance? clojure.lang.IEditableCollection to)
+     (with-meta (persistent! (stream-reduce! conj! (transient to) stream)) (meta to))
+     (stream-reduce! conj to stream)))
+  ([to xform ^java.util.stream.BaseStream stream]
+   (if (instance? clojure.lang.IEditableCollection to)
+     (with-meta (persistent! (stream-transduce! xform conj! (transient to) stream)) (meta to))
+     (stream-transduce! xform conj to stream))))
+
 (defmacro ^:private when-class [class-name & body]
   `(try
      (Class/forName ^String ~class-name)
@@ -6887,6 +6924,7 @@ fails, attempts to require sym's namespace and retries."
    :added "1.11"}
   ^java.util.UUID [] (java.util.UUID/randomUUID))
 
+;; redefine reduce with internal-reduce
 (defn reduce
   "f should be a function of 2 arguments. If val is not supplied,
   returns the result of applying f to the first 2 items in coll, then

src/clj/clojure/core/protocols.clj

@@ -30,27 +30,26 @@
      (let [s (seq coll)]
        (internal-reduce s f val))))
 
-(defn- iter-reduce
-  ([^java.lang.Iterable coll f]
-   (let [iter (.iterator coll)]
+;; mutates the iterator, respects reduced
+(defn iterator-reduce!
+  ([^java.util.Iterator iter f]
+   (if (.hasNext iter)
+     (iterator-reduce! iter f (.next iter))
+     (f)))
+  ([^java.util.Iterator iter f val]
+   (loop [ret val]
      (if (.hasNext iter)
-       (loop [ret (.next iter)]
-         (if (.hasNext iter)
-           (let [ret (f ret (.next iter))]
-             (if (reduced? ret)
-               @ret
-               (recur ret)))
-           ret))
-       (f))))
-  ([^java.lang.Iterable coll f val]
-   (let [iter (.iterator coll)]
-     (loop [ret val]
-       (if (.hasNext iter)
-         (let [ret (f ret (.next iter))]
-           (if (reduced? ret)
-             @ret
-             (recur ret)))
-         ret)))))
+       (let [ret (f ret (.next iter))]
+         (if (reduced? ret)
+           @ret
+           (recur ret)))
+       ret))))
+
+(defn- iter-reduce
+  ([^Iterable coll f]
+   (iterator-reduce! (.iterator coll) f))
+  ([^Iterable coll f val]
+   (iterator-reduce! (.iterator coll) f val)))
 
 (defn- naive-seq-reduce
   "Reduces a seq, ignoring any opportunities to switch to a more

test/clojure/test_clojure/streams.clj

@@ -0,0 +1,103 @@
+;   Copyright (c) Rich Hickey. All rights reserved.
+;   The use and distribution terms for this software are covered by the
+;   Eclipse Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php)
+;   which can be found in the file epl-v10.html at the root of this distribution.
+;   By using this software in any fashion, you are agreeing to be bound by
+;   the terms of this license.
+;   You must not remove this notice, or any other, from this software.
+
+(ns clojure.test-clojure.streams
+  (:use clojure.test)
+  (:import [java.util.stream Stream LongStream]
+           [java.util.function Consumer Predicate Supplier]))
+
+(deftest test-stream-reduce!
+  (is (= :only-val (stream-reduce! + (.stream [:only-val]))))
+  (is (= 0 (stream-reduce! + (.stream []))))
+  (is (= 5 (stream-reduce! + (.stream [1 4]))))
+  (is (= 6 (stream-reduce! + (.stream [1 2 3]))))
+  (is (= [2 3 4]
+         (stream-reduce! (fn [v i] (conj v (inc i)))
+                         []
+                         (.stream [1 2 3]))))
+
+  (is (= 15 (stream-reduce! + (LongStream/rangeClosed 1 5))))
+
+  (is (= 9 (stream-reduce! + (-> (Stream/of (to-array [1 2 3 4 5]))
+                                 (.filter (reify Predicate (test [_ v] (odd? v))))))))
+  (is (= {:a 1}
+           (meta
+            (stream-reduce! (fn [v i] (conj v (inc i)))
+                            (with-meta [] {:a 1})
+                            (.stream [1 2 3]))))))
+
+(deftest test-stream-reduced!
+  (is (= 5 (stream-reduce!
+            (fn [x y] (reduced (+ x y)))
+            (.stream [1 4]))))
+
+    (is (= 45 (stream-reduce!
+               (fn [acc v]
+                 (if (= v 10)
+                   (reduced acc)
+                   (+ acc v)))
+               (LongStream/rangeClosed 1 10)))))
+
+(deftest stream-seq!-test
+  (let [none (.stream [])
+        one  (Stream/of "a")
+        n    (.stream ["a" "b" "c"])
+	inf  (Stream/generate (reify Supplier (get [_] 42)))
+        st   (stream-seq! one)
+        l100 (LongStream/range 0 100)]
+    (is (empty? (map identity (stream-seq! none))))
+    (is (seq? st))
+    (is (= ["a"] (map identity st)))
+    (is (= ["a" "b" "c"] (map identity (stream-seq! n))))
+    (is (= [42 42 42 42 42] (take 5 (stream-seq! inf))))
+    (is (= 4950 (reduce + (stream-seq! l100)))) 4950))
+
+(deftest stream-transduce!-test
+  (let [xf (comp (filter odd?) (take 5))]
+    (let [st (Stream/of (to-array [1 2 3 4 5 6 7 8 9]))]
+      (is (= [1 3 5 7 9] (stream-transduce! xf conj st))))
+
+    (let [inf  (Stream/generate (reify Supplier (get [_] 0)))]
+      (is (empty? (stream-transduce! xf conj (.limit inf 50)))))
+
+    (let [inf  (Stream/generate (reify Supplier (get [_] 43)))]
+      (is (= [43 43 43 43 43] (stream-transduce! xf conj (.limit inf 50)))))
+
+    (let [inf  (Stream/generate (reify Supplier (get [_] 43)))]
+      (is (= 215 (stream-transduce! xf + (.limit inf 50)))))
+
+    (let [inf  (Stream/generate (reify Supplier (get [_] 43)))]
+      (is (= 315 (stream-transduce! xf + 100 (.limit inf 50)))))
+
+    (let [inf  (Stream/generate (reify Supplier (get [_] 43)))]
+      (is (= "4343434343" (stream-transduce! xf str (.limit inf 50)))))))
+
+(deftest stream-into!-test
+  (let [none (.stream [])
+        one  (Stream/of "a")
+        n    (.stream ["a" "b" "c"])
+        inf  (Stream/generate (reify Supplier (get [_] 42)))
+        par  (-> (LongStream/rangeClosed 1 10) .boxed .parallel)
+        xf   (comp (map #(+ 2 %)) (filter odd?))
+        par2 (-> (LongStream/rangeClosed 1 10) .boxed .parallel)]
+    (is (empty? (stream-into! [] none)))
+    (is (= ["a"] (stream-into! [] one)))
+    (is (= ["a" "b" "c"] (stream-into! [] n)))
+    (is (= [42 42 42 42 42]
+           (stream-into! [] (.limit inf 5))))
+    (is (= [1 2 3 4 5 6 7 8 9 10]
+           (stream-into! [] par)))
+    (is (= {:a 1}
+           (meta
+            (stream-into! (with-meta [] {:a 1})
+                          (.stream [1 2 3])))))
+    (is (= {:a 1}
+           (meta
+            (stream-into! (with-meta clojure.lang.PersistentQueue/EMPTY {:a 1})
+                          (.stream [1 2 3])))))
+    (is (= [-1 -2 3 5 7 9 11] (stream-into! [-1 -2] xf par2)))))