             [runbld.util.date :as date]
             [runbld.util.io :as io]))
 
+(def empty-metrics
+  {:ordinal
+   {:global 0
+    :stdout 0
+    :stderr 0}
+   :bytes
+   {:global 0
+    :stdout 0
+    :stderr 0}})
+
+(s/defn capture-metrics
+  [stream :- s/Keyword
+   metrics :- clojure.lang.Ref]
+  (fn [line]
+    (let [line-bytes (inc (count (.getBytes line)))]
+      (alter metrics update-in [:ordinal :global] inc)
+      (alter metrics update-in [:ordinal stream] inc)
+      (alter metrics update-in [:bytes :global] + line-bytes)
+      (alter metrics update-in [:bytes stream] + line-bytes))))
+
 (s/defn capture-ordinal
   [ord :- clojure.lang.Atom]
   (fn [line]

    es-opts    :- OptsElasticsearch
    env        :- {s/Str s/Str}]
   (let [dir (io/abspath-file cwd)
         global-ordinal (atom 0)
         out-file (io/prepend-path dir stdout)
         out-ordinal (atom 0)
         out-bytes (atom 0)
         err-file (io/prepend-path dir stderr)
         err-ordinal (atom 0)
         err-bytes (atom 0)
         result
         (with-open [outw (java.io.PrintWriter. out-file)
                     errw (java.io.PrintWriter. err-file)]
           (let [processors
                 {:out [(capture-ordinal global-ordinal)
                        (capture-ordinal out-ordinal)
                        (capture-bytes out-bytes)
                        (capture-to-writer *out*)
                        (capture-to-writer outw)
                        (capture-to-es es-opts
                                       build-id
                                       :stdout
                                       global-ordinal
                                       out-ordinal)]
                  :err [(capture-ordinal global-ordinal)
                        (capture-ordinal err-ordinal)
                        (capture-bytes err-bytes)
                        (capture-to-writer *err*)
                        (capture-to-writer errw)
                        (capture-to-es es-opts
                                       build-id
                                       :stderr
                                       global-ordinal
                                       err-ordinal)]}]
             (exec program args scriptfile cwd env processors)))
         out-file-bytes (count (slurp out-file))
         err-file-bytes (count (slurp err-file))]
     (store/after-log es-opts)
     (merge
      result
      {:out-bytes @out-bytes
       :err-bytes @err-bytes
       :out-file (str out-file)
       :err-file (str err-file)
       :out-file-bytes out-file-bytes
       :err-file-bytes err-file-bytes
       :out-accuracy (data/scaled-percent
                      out-file-bytes @out-bytes)
       :err-accuracy (data/scaled-percent
                      err-file-bytes @err-bytes)}))))

(s/defn run :- {(s/required-key :opts) MainOpts
                (s/required-key :process-result) ProcessResult}

      (flush))))

(s/defn capture-to-es
  ([opts    :- OptsElasticsearch
    id      :- s/Str
    stream  :- s/Keyword
    metrics :- clojure.lang.Ref]
   (fn [line]
     (let [doc {:build-id id
                :stream (name stream)
                :time (date/ms-to-iso)
                :log line
                :ordinal {:global (get-in @metrics [:ordinal :global])
                          :stream (get-in @metrics [:ordinal stream])}}]
       (store/save-log! opts doc)))))

(s/defn spit-stream :- s/Keyword
  [input        :- java.io.InputStream
   processors   :- [clojure.lang.IFn]]
  (doseq [line (line-seq (io/reader input))]
    ;; dosync: in order for the global ordinal to order in accordance
    ;; with time, metrics calculation can only happen one stream at a
    ;; time
    (dosync
     (doseq [processor processors]
       (processor line))))
  :done)

(s/defn spit-process :- [java.util.concurrent.Future]
