 (defn ^{:added "0.6.0"} start-env
   "Starts a development environment using an in-memory implementation ZooKeeper."
   ([env-config]
-     (start-env env-config {:monitoring :no-op}))
+   (start-env env-config {:monitoring :no-op}))
   ([env-config monitoring-config]
    (validator/validate-env-config env-config)
    (component/start (system/onyx-development-env env-config monitoring-config))))

   Task lifecycle APIs for closing tasks are invoked. This job is never again scheduled
   for execution."
  ([peer-config job-id]
     (kill-job peer-config job-id {:monitoring :no-op}))
  ([peer-config job-id monitoring-config]
     (validator/validate-peer-config peer-config)
     (when (nil? job-id)
       (throw (ex-info "Invalid job id" {:job-id job-id})))
     (let [client (component/start (system/onyx-client peer-config monitoring-config))
           entry (create-log-entry :kill-job {:job (validator/coerce-uuid job-id)})]
       (extensions/write-log-entry (:log client) entry)
       (component/stop client)
       true)))

(defn ^{:added "0.6.0"} subscribe-to-log
  "Sends all events from the log to the provided core.async channel.

   a 3rd argument - a monitoring configuration map. See the User Guide
   for details."
  ([n peer-group]
   (start-peers n peer-group {:monitoring :no-op}))
  ([n {:keys [config] :as peer-group} monitoring-config]
   (when-not (= (type peer-group) onyx.system.OnyxPeerGroup)
     (throw (Exception. (str "start-peers must supplied with a peer-group not a " (type peer-group)))))
   (validator/validate-java-version)
   (doall
    (map
     (fn [_]
       (let [v-peer (system/onyx-peer peer-group monitoring-config)
             live (component/start v-peer)
             shutdown-ch (chan 1)
             ack-ch (chan)
             started-peer (atom live)]
         {:peer-lifecycle (future (peer-lifecycle started-peer config shutdown-ch ack-ch))
          :started-peer started-peer
          :shutdown-ch shutdown-ch
          :ack-ch ack-ch}))
     (range n)))))

(defn ^{:added "0.6.0"} shutdown-peer
  "Shuts down the virtual peer, which releases all of its resources
