 (ns parkour.inspect.mapreduce
   {:private true}
-  (:require [parkour (mapreduce :as mr) (fs :as fs)]
+  (:require [clojure.core.protocols :as ccp]
+            [parkour (mapreduce :as mr) (fs :as fs) (conf :as conf)]
             [parkour.util :refer [ignore-errors doto-let returning]])
   (:import [java.io Closeable]
            [clojure.lang Seqable]

    (returning nil (.close rr'))
    (returning rr' (close-rr rr))))

(defn records-seqable
  [conf f klass & paths]
  (let [job (doto-let [job (mr/job conf)]
              (doseq [path paths :let [path (fs/path path)]]
                (FileInputFormat/addInputPath job path)))
        conf (.getConfiguration job)
        ^InputFormat ifi (ReflectionUtils/newInstance klass conf)
        splits (seq (.getSplits ifi job))
        closer (agent ::initial)]
    (reify
      Closeable (close [_] (send closer close-rr))
      Seqable
      (seq [_]
        ((fn step [^RecordReader rr splits]

                           (.initialize split tac))]
                  (returning (step rr splits)
                    (send closer update-rr rr)))))))
         nil splits))

      ccp/CollReduce
      (coll-reduce [this f1] (ccp/coll-reduce this f1 (f1)))
      (coll-reduce [_ f1 init]
        (let [rra (atom ::initial)]
          (try
            (loop [acc init, splits splits, rr nil]
              (if (and rr (.nextKeyValue ^RecordReader rr))
                (let [acc (->> rr rr-tuple f (f1 acc))]
                  (if (reduced? acc)
                    @acc
                    (recur acc splits rr)))
                (if (empty? splits)
                  acc
                  (let [split (first splits), splits (rest splits)
                        tac (new-tac conf (TaskAttemptID.)),
                        rr (doto (.createRecordReader ifi split tac)
                             (.initialize split tac))]
                    (recur acc splits (swap! rra update-rr rr))))))
            (finally
              (close-rr @rra))))))))
