 
 (def out-chan (chan (inc chan-size)))
 
-(defmethod l-ext/inject-lifecycle-resources :in
-  [_ _] {:core.async/chan in-chan})
+(defn inject-in-ch [event lifecycle]
+  {:core.async/chan in-chan})
 
-(defmethod l-ext/inject-lifecycle-resources :out
-  [_ _] {:core.async/chan out-chan})
+(defn inject-out-ch [event lifecycle]
+  {:core.async/chan out-chan})
+
+(def in-calls
+  {:lifecycle/before-task :onyx.peer.fault-tolerance/inject-in-ch})
+
+(def out-calls
+  {:lifecycle/before-task :onyx.peer.fault-tolerance/inject-out-ch})
+
+(def lifecycles
+  [{:lifecycle/task :in
+    :lifecycle/calls :onyx.peer.fault-tolerance/in-calls}
+   {:lifecycle/task :in
+    :lifecycle/calls :onyx.plugin.core-async/reader-calls}
+   {:lifecycle/task :out
+    :lifecycle/calls :onyx.peer.fault-tolerance/out-calls}
+   {:lifecycle/task :out
+    :lifecycle/calls :onyx.plugin.core-async/writer-calls}])
 
 (def incomplete 
   (atom #{}))


(def n-messages-total 100000)

(try
  (helper-env/add-peers test-env 9)
  (let [load-data-ch (thread 
                       (loop [ls (repeatedly n-messages-total (fn [] {:id (java.util.UUID/randomUUID)}))]
                         (when-let [segment (first ls)]
                           (swap! incomplete conj (int-2 (int-1 segment)))
                           (>!! in-chan segment)
                           (recur (rest ls))))
                       (>!! in-chan :done)
                       (close! in-chan))

        job-id (:job-id (helper-env/run-job test-env 
                                            {:catalog catalog
                                             :workflow workflow
                                             :task-scheduler :onyx.task-scheduler/balanced}))
        _ (Thread/sleep 4000)
        task-peers (-> test-env
                       :replica
                       deref
                       :allocations
                       (get job-id))

        task-ids (->> task-peers
                      keys
                      (map (partial extensions/read-chunk (:log (:env test-env)) :task))
                      (map (juxt :name :id))
                      (into {}))

        non-input-task-ids (set (vals (dissoc task-ids :in)))

        remove-completed-ch (go-loop []
                                     (let [v (<!! out-chan)]
                                       (when (and v (not= v :done)) 
                                         (swap! incomplete disj v)
                                         (when (or (nil? (:a v))
                                                   (nil? (:b v)))
                                           (swap! bad-values conj v))
                                         (recur))))
        kill-ch (chan 1)
        mess-with-peers-ch (go-loop []
                                    (let [; average the the full circuit worth of timeouts
                                          chaos-kill-ms (rand-int (* 3 batch-timeout 2))
                                          [v ch] (alts!! [(timeout chaos-kill-ms) kill-ch])]
                                      (when-not (= ch kill-ch) 
                                        (try 
                                          (when-let [non-input-peers (->> (get (:allocations @(:replica test-env)) job-id)
                                                                          (filter (fn [[task-id peers]]
                                                                                    (and (> (count peers) 1)
                                                                                         (non-input-task-ids task-id))))
                                                                          vals
                                                                          flatten
                                                                          seq)]
                                            (when-let [peer-val (helper-env/lookup-peer test-env (rand-nth non-input-peers))] 
                                              (println "Killing a peer")
                                              (helper-env/remove-peer test-env peer-val)
                                              (helper-env/add-peers test-env 1)))
                                          (catch Exception e
                                            (println "FAILURE MANAGING: " e)))
                                        (recur))))]

    (<!! remove-completed-ch)
    (fact @incomplete => #{})
    (let [after-done (loop [after-done #{}]
                       (let [[v ch] (alts!! [out-chan (timeout 10000)])]
                         (if v 
                           (recur (conj after-done v))
                           after-done)))]
      (fact after-done => #{}))
      (fact @bad-values => #{})
    (close! kill-ch)
    #_(<!! mess-with-peers-ch))
  (finally 
    (component/stop test-env)))

  (:require [onyx.helper-env :as helper-env :refer [->TestEnv map->TestEnv]]
            [clojure.core.async :refer [chan >!! <!! close! go-loop timeout thread alts!! sliding-buffer]]
            [midje.sweet :refer :all]
            [onyx.extensions :as extensions]
            [com.stuartsierra.component :as component]
            [onyx.plugin.core-async :refer [take-segments!]]
