 (ns onyx.peer.kw-grouping-test
-  (:require [com.stuartsierra.component :as component]
-            [onyx.system :refer [onyx-development-env]]
+  (:require [clojure.core.async :refer [chan >!! <!! close! sliding-buffer]]
+            [midje.sweet :refer :all]
             [onyx.peer.task-lifecycle-extensions :as l-ext]
-            [taoensso.timbre :refer [info]]
-            [midje.sweet :refer :all]
-            [onyx.queue.hornetq-utils :as hq-util]
+            [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.kw-grouping-test/sum-balance

  (swap! state (fn [v] (assoc v name (+ (get v name 0) amount))))
  [])

(def workflow
  [[:in :sum-balance]
   [:sum-balance :out]])

(def catalog
  [{:onyx/name :in
    :onyx/ident :core.async/read-from-chan
    :onyx/type :input
    :onyx/medium :core.async
    :onyx/batch-size 40
    :onyx/max-peers 1
    :onyx/doc "Reads segments from a core.async channel"}

   {:onyx/name :sum-balance
    :onyx/ident :onyx.peer.kw-grouping-test/sum-balance
    :onyx/fn :onyx.peer.kw-grouping-test/sum-balance
    :onyx/type :function
    :onyx/group-by-key :name
    :onyx/batch-size 40}

   {:onyx/name :out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size 40
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}])

(def size 3)

(def data
  (concat
