    (map (fn [_] {:name "Eusebia" :amount 10}) (range size))
    (map (fn [_] {:name "Fletcher" :amount 10}) (range size))))
 
-(hq-util/write-and-cap! hq-config in-queue data 100)
+(doseq [x data]
+  (>!! in-chan x))
 
-(def v-peers (onyx.api/start-peers! 6 peer-config))
+(>!! in-chan :done)
+(close! in-chan)
 
-(onyx.api/submit-job peer-config
-                     {:catalog catalog :workflow workflow
-                      :task-scheduler :onyx.task-scheduler/round-robin})
+(def v-peers (onyx.api/start-peers 3 peer-config))
 
-(def results (hq-util/consume-queue! hq-config out-queue 1))
+(onyx.api/submit-job
+ peer-config
+ {:catalog catalog :workflow workflow
+  :task-scheduler :onyx.task-scheduler/round-robin})
 
-(doseq [v-peer v-peers]
-  (onyx.api/shutdown-peer v-peer))
-
-(onyx.api/shutdown-env env)
+(def results (take-segments! out-chan))
 
 (def out-val @output)
 


(def output (atom []))

(def hq-config {"host" (:host (:non-clustered (:hornetq config)))
                "port" (:port (:non-clustered (:hornetq config)))})

(def in-queue (str (java.util.UUID/randomUUID)))

(def out-queue (str (java.util.UUID/randomUUID)))

(hq-util/create-queue! hq-config in-queue)
(hq-util/create-queue! hq-config out-queue)

(defmethod l-ext/inject-lifecycle-resources
  :onyx.peer.fn-grouping-test/sum-balance

(ns onyx.peer.fn-grouping-test
  (:require [clojure.core.async :refer [chan >!! <!! close! sliding-buffer]]
            [midje.sweet :refer :all]
            [onyx.peer.task-lifecycle-extensions :as l-ext]
            [onyx.plugin.core-async :refer [take-segments!]]
            [onyx.api]))

(def id (java.util.UUID/randomUUID))
