 
 (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))
+(when false
+  (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))
+          job-id (:job-id (helper-env/run-job test-env 
+                                              {:catalog catalog
+                                               :workflow workflow
+                                               :lifecycles lifecycles
+                                               :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 {}))
+          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)))
+          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 => #{}))
+          remove-completed-ch (go-loop []
+                                       (let [v (<!! out-chan)]
+                                         (println "got " v)
+                                         (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)
+          chaos-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 (and (= ch kill-ch) v) 
+                                (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)
+      (println "All done")
+      (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)))
+      (close! kill-ch)
+      #_(<!! mess-with-peers-ch))
+    (finally 
+      (component/stop test-env))))


(def out-chan (chan (inc chan-size)))

(defmethod l-ext/inject-lifecycle-resources :in
  [_ _] {:core.async/chan in-chan})

(defmethod l-ext/inject-lifecycle-resources :out
  [_ _] {:core.async/chan out-chan})

(def incomplete 
  (atom #{}))

  (: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!]]
