 
 (def output (atom []))
 
-(def hq-config {"host" (:host (:non-clustered (:hornetq config)))
-                "port" (:port (:non-clustered (:hornetq config)))})
+(def in-chan (chan 1000))
 
-(def in-queue (str (java.util.UUID/randomUUID)))
+(def out-chan (chan (sliding-buffer 1000)))
 
-(def out-queue (str (java.util.UUID/randomUUID)))
+(defmethod l-ext/inject-lifecycle-resources :in
+  [_ _] {:core.async/chan in-chan})
 
-(hq-util/create-queue! hq-config in-queue)
-(hq-util/create-queue! hq-config out-queue)
+(defmethod l-ext/inject-lifecycle-resources :out
+  [_ _] {:core.async/chan out-chan})
 
 (defmethod l-ext/inject-lifecycle-resources
   :onyx.peer.kw-grouping-test/sum-balance

(ns onyx.peer.kw-grouping-test
  (:require [com.stuartsierra.component :as component]
            [onyx.system :refer [onyx-development-env]]
            [onyx.peer.task-lifecycle-extensions :as l-ext]
            [taoensso.timbre :refer [info]]
            [midje.sweet :refer :all]
            [onyx.queue.hornetq-utils :as hq-util]
            [onyx.api]))

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

  (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
