 (ns onyx.peer.flow-test
-  (:require [midje.sweet :refer :all]
-            [onyx.system :refer [onyx-development-env]]
-            [onyx.queue.hornetq-utils :as hq-util]
+  (: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]
             [onyx.api]))
 
 (def id (java.util.UUID/randomUUID))

    :onyx/batch-size batch-size}

   {:onyx/name :children-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name children-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :adults-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name adults-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :athletes-wa-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name athletes-wa-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :everyone-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name everyone-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :red-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name red-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :blue-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name blue-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :green-out
    :onyx/ident :hornetq/write-segments
    :onyx/type :output
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name green-out-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}])

(def workflow
  [[:people-in :process-children]


(def echo 1000)

(def people-in-chan (chan (inc n-messages)))

(def colors-in-chan (chan (inc n-messages)))

(def children-out-chan (chan (sliding-buffer (inc n-messages))))

(def adults-out-chan (chan (sliding-buffer (inc n-messages))))

(def athletes-wa-out-chan (chan (sliding-buffer (inc n-messages))))

(def everyone-out-chan (chan (sliding-buffer (inc n-messages))))

(def red-out-chan (chan (sliding-buffer (inc n-messages))))

(def blue-out-chan (chan (sliding-buffer (inc n-messages))))

(def green-out-chan (chan (sliding-buffer (inc n-messages))))

(doseq [x [{:age 24 :job "athlete" :location "Washington"}
           {:age 17 :job "programmer" :location "Washington"}
           {:age 18 :job "mechanic" :location "Vermont"}
           {:age 13 :job "student" :location "Maine"}
           {:age 42 :job "doctor" :location "Florida"}
           {:age 64 :job "athlete" :location "Pennsylvania"}
           {:age 35 :job "bus driver" :location "Texas"}
           {:age 50 :job "lawyer" :location "California"}
           {:age 25 :job "psychologist" :location "Washington"}]]
  (>!! people-in-chan x))

(>!! people-in-chan :done)

(doseq [x [{:color "red" :extra-key "Some extra context for the predicates"}
           {:color "blue" :extra-key "Some extra context for the predicates"}
           {:color "white" :extra-key "Some extra context for the predicates"}
           {:color "green" :extra-key "Some extra context for the predicates"}
           {:color "orange" :extra-key "Some extra context for the predicates"}
           {:color "black" :extra-key "Some extra context for the predicates"}
           {:color "purple" :extra-key "Some extra context for the predicates"}
           {:color "cyan" :extra-key "Some extra context for the predicates"}
           {:color "yellow" :extra-key "Some extra context for the predicates"}]]
  (>!! colors-in-chan x))

(>!! colors-in-chan :done)

(close! people-in-chan)
(close! colors-in-chan)

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

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

   {:onyx/name :process-children
    :onyx/fn :onyx.peer.flow-test/process-children
