 (ns onyx.peer.fault-tolerance
   (:require [onyx.helper-env :as helper-env :refer [->TestEnv map->TestEnv]]
-            [clojure.core.async :refer [chan >!! <!! close! go-loop timeout alts!! sliding-buffer]]
+            [clojure.core.async :refer [chan >!! <!! close! go-loop timeout thread alts!! sliding-buffer]]
             [midje.sweet :refer :all]
             [onyx.peer.task-lifecycle-extensions :as l-ext]
             [onyx.extensions :as extensions]

                                         (swap! incomplete disj v)
                                         (recur))))
        kill-ch (chan 1)
        chaos-kill-ms 1000
        mess-with-peers-ch (go-loop []
                                    (let [[v ch] (alts!! [(timeout chaos-kill-ms) kill-ch])]
                                      (when-not (= ch kill-ch) 
                                        (try 
                                          (let [rand-non-input-peer (->> (get (:allocations @(:replica test-env)) job-id)
                                                                         (filter (fn [[task-id peers]]
                                                                                   (and (> (count peers) 1)
                                                                                        (non-input-task-ids task-id))))
                                                                         vals
                                                                         flatten
                                                                         rand-nth)
                                                peer-val (helper-env/lookup-peer test-env rand-non-input-peer)]
                                            (when peer-val
                                              (helper-env/remove-peer test-env peer-val)
                                              (println "Removed peer " rand-non-input-peer)
                                              (helper-env/add-peers test-env 1)))
                                          (catch Exception e
                                            (println "FAILURE MANAGING: " e)))


(def process-middle identity)

(def batch-timeout 500)

(def catalog
  [{:onyx/name :in
    :onyx/ident :core.async/read-from-chan
    :onyx/type :input
    :onyx/medium :core.async
    :onyx/batch-timeout batch-timeout
    :onyx/max-peers 1
    :onyx/batch-size batch-size
    :onyx/doc "Reads segments from a core.async channel"}

   {:onyx/name :intermediate-1
    :onyx/fn :onyx.peer.fault-tolerance/process-middle
    :onyx/batch-timeout batch-timeout
    :onyx/type :function
    :onyx/batch-size batch-size}

   {:onyx/name :intermediate-2
    :onyx/fn :onyx.peer.fault-tolerance/process-middle
    :onyx/type :function
    :onyx/batch-timeout batch-timeout
    :onyx/batch-size batch-size}

   {:onyx/name :out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/batch-timeout batch-timeout
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/doc "Writes segments to a core.async channel"}])
