 (ns onyx.peer.fn-grouping-test
-  (:require [onyx.peer.task-lifecycle-extensions :as l-ext]
-            [taoensso.timbre :refer [info]]
+  (:require [clojure.core.async :refer [chan >!! <!! close! sliding-buffer]]
             [midje.sweet :refer :all]
-            [onyx.queue.hornetq-utils :as hq-util]
+            [onyx.peer.task-lifecycle-extensions :as l-ext]
+            [onyx.plugin.core-async :refer [take-segments!]]
             [onyx.api]))
 
 (def id (java.util.UUID/randomUUID))


(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

   (map (fn [_] {:name "Eusebia" :amount 10}) (range size))
   (map (fn [_] {:name "Fletcher" :amount 10}) (range size))))

(doseq [x data]
  (>!! in-chan x))

(>!! in-chan :done)
(close! in-chan)

(def v-peers (onyx.api/start-peers 3 peer-config))

(onyx.api/submit-job
 peer-config
 {:catalog catalog :workflow workflow
  :task-scheduler :onyx.task-scheduler/round-robin})

(def results (take-segments! out-chan))

(def out-val @output)

