gistfile1.clj

(defn with-env-peers [env-config peer-config n-peers f]
  (let [env (onyx.api/start-env env-config)
        v-peers (onyx.api/start-peers! n-peers peer-config)]
    (try
      (f)
      (finally
        (doseq [v-peer v-peers]
          (onyx.api/shutdown-peer v-peer))
        (onyx.api/shutdown-env env)))))

(with-env-peers
  env-config peer-config 3
  (fn []
    (onyx.api/submit-job
      peer-config
      {:catalog catalog :workflow workflow
       :task-scheduler :onyx.task-scheduler/round-robin})
    (let [results (doall (repeatedly n-messages (fn [] (<!! out-chan))))
          expected (set (map (fn [x] {:n (inc x)}) (range n-messages)))]
      (fact (set results) => expected)
      ;  (fact (last results) => :done)
      )))