     :hornetq/queue-name in-queue
     :hornetq/host hornetq-host
     :hornetq/port hornetq-port
-    :hornetq/batch-size 2}
+    :onyx/batch-size 2}
 
    {:onyx/name :to-datom
     :onyx/fn :onyx.plugin.output-test/to-datom
-    :onyx/type :transformer
+    :onyx/type :function
     :onyx/consumption :concurrent
     :onyx/batch-size 2}
    
-   {:onyx/name :output
+   {:onyx/name :out
     :onyx/ident :datomic/commit-tx
     :onyx/type :output
     :onyx/medium :datomic


(hq-utils/write-and-cap! hq-config in-queue people 1)

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

(def coord-opts {:hornetq/mode :vm
                 :hornetq/server? true
                 :hornetq.server/type :vm
                 :zookeeper/address "127.0.0.1:2185"
                 :zookeeper/server? true
                 :zookeeper.server/port 2185
                 :onyx/id id
                 :onyx.coordinator/revoke-delay 5000})

(def peer-opts
  {:hornetq/mode :vm
   :zookeeper/address "127.0.0.1:2185"
   :onyx/id id})

(def workflow {:input {:to-datom :output}})

(def catalog
  [{:onyx/name :input
    :onyx/ident :hornetq/read-segments
    :onyx/type :input
    :onyx/medium :hornetq

  {:datoms [{:db/id (d/tempid :com.mdrogalis/people)
             :user/name name}]})

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

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

(doseq [_ (range (count people))]
  (.take tx-queue))
