   (swap! state (fn [v] (assoc v name (+ (get v name 0) amount))))
   [])
 
-(def workflow {:in {:sum-balance :out}})
+(def workflow
+  [[:in :sum-balance]
+   [:sum-balance :out]])
 
 (def catalog
   [{:onyx/name :in
-    :onyx/ident :hornetq/read-segments
+    :onyx/ident :core.async/read-from-chan
     :onyx/type :input
-    :onyx/medium :hornetq
-    :hornetq/queue-name in-queue
-    :hornetq/host (:host (:non-clustered (:hornetq config)))
-    :hornetq/port (:port (:non-clustered (:hornetq config)))
-    :onyx/batch-size 1000}
+    :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 1000}
+    :onyx/batch-size 40}
 
    {:onyx/name :out
-    :onyx/ident :hornetq/write-segments
+    :onyx/ident :core.async/write-to-chan
     :onyx/type :output
-    :onyx/medium :hornetq
-    :hornetq/queue-name out-queue
-    :hornetq/host (:host (:non-clustered (:hornetq config)))
-    :hornetq/port (:port (:non-clustered (:hornetq config)))
-    :onyx/batch-size 1000}])
+    :onyx/medium :core.async
+    :onyx/batch-size 40
+    :onyx/max-peers 1
+    :onyx/doc "Writes segments to a core.async channel"}])
 
-(def size 3000)
+(def size 3)
 
 (def data
   (concat

   (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)

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

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

(def results (hq-util/consume-queue! hq-config out-queue 1))

(doseq [v-peer v-peers]
  (onyx.api/shutdown-peer v-peer))


(def output (atom []))

(def in-chan (chan 1000))

(def out-chan (chan (sliding-buffer 1000)))

(defmethod l-ext/inject-lifecycle-resources :in
  [_ _] {:core.async/chan in-chan})

(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
