 (defn group-by-name [{:keys [name]}]
   name)
 
-(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.fn-grouping-test/sum-balance
     :onyx/fn :onyx.peer.fn-grouping-test/sum-balance
     :onyx/type :function
     :onyx/group-by-fn :onyx.peer.fn-grouping-test/group-by-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 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))
