 
 (def ch (chan n-peers))
 
-(def subscription (onyx.api/subscribe-to-log peer-config ch))
-
-(def log (:log (:env subscription)))
-
 (def replica
   (playback-log (:log env) (extensions/subscribe-to-log (:log env) ch) ch 2000))
 
-(onyx.api/shutdown-env (:env subscription))
-
-(fact "peers balanced on 1 jobs" 
+(fact "peers balanced on 1 job"
       (into #{} (map count (mapcat vals (vals (:allocations replica))))) 
       =>
       #{1 2})
+
+(onyx.api/shutdown-env (:env subscription))
 
 (doseq [v-peer v-peers]
   (onyx.api/shutdown-peer v-peer))

    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/doc "Writes segments to a core.async channel"}])

(def in-chan (chan 100))

    :onyx/type :input
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/batch-timeout 2000
    :onyx/doc "Reads segments from a core.async channel"}

   {:onyx/name :b
    :onyx/fn :onyx.log.one-job-test/my-inc
    :onyx/type :function
    :onyx/batch-timeout 2000
    :onyx/batch-size batch-size}

   {:onyx/name :c
