snap

project.clj

@@ -1,4 +1,4 @@
-(defproject cnuernber/dtype-next "7.10"
+(defproject cnuernber/dtype-next "7.11-SNAPSHOT"
   :description "A Clojure library designed to aid in the implementation of high performance algorithms and systems."
   :url "http://github.com/cnuernber/dtype-next"
   :license {:name "EPL-2.0"

src/tech/v3/parallel/for.clj

@@ -1,5 +1,6 @@
 (ns tech.v3.parallel.for
   "Serial and parallel iteration strategies across iterators and index spaces."
+  (:require [tech.v3.datatype.errors :as errors])
   (:import [java.util.concurrent ForkJoinPool Callable Future ForkJoinTask]
            [java.util ArrayDeque PriorityQueue Comparator Spliterator Iterator
             ArrayList List]
@@ -225,10 +226,26 @@
       (if (convertible-to-iterator? item)
         (doiter
          value item
-         (.accept local-consumer value))))
+         (.accept local-consumer value))
+        (errors/throwf "Item is not iterable thus it cannot be consumed")))
     consumer))
 
 
+(defn indexed-consume!
+  "Consume (terminate) a sequence or stream.  If the stream is parallel
+  then the consumer had better be threadsafe.  Consumer in this case is
+  expected to be an clojure IFn and it will receive two arguments, a long
+  and the value."
+  [consumer item]
+  (let [iter (->iterator item)]
+    (loop [continue? (.hasNext iter)
+           idx 0]
+      (when continue?
+        (consumer idx (.next iter))
+        (recur (.hasNext iter) (unchecked-inc idx)))))
+  consumer)
+
+
 (defn spliterator-map-reduce
   "Given a spliterator, a function from spliterator to scalar and a
   reduction function do an efficient parallelized reduction."