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


(def echo 1000)

(def people-in-queue (str (java.util.UUID/randomUUID)))

(def colors-in-queue (str (java.util.UUID/randomUUID)))

(def children-out-queue (str (java.util.UUID/randomUUID)))

(def adults-out-queue (str (java.util.UUID/randomUUID)))

(def athletes-wa-out-queue (str (java.util.UUID/randomUUID)))

(def everyone-out-queue (str (java.util.UUID/randomUUID)))

(def red-out-queue (str (java.util.UUID/randomUUID)))

(def blue-out-queue (str (java.util.UUID/randomUUID)))

(def green-out-queue (str (java.util.UUID/randomUUID)))

(def hq-config {"host" (:host (:non-clustered (:hornetq config)))
                "port" (:port (:non-clustered (:hornetq config)))})

(hq-util/create-queue! hq-config people-in-queue)
(hq-util/create-queue! hq-config colors-in-queue)
(hq-util/create-queue! hq-config children-out-queue)
(hq-util/create-queue! hq-config adults-out-queue)
(hq-util/create-queue! hq-config athletes-wa-out-queue)
(hq-util/create-queue! hq-config everyone-out-queue)
(hq-util/create-queue! hq-config red-out-queue)
(hq-util/create-queue! hq-config blue-out-queue)
(hq-util/create-queue! hq-config green-out-queue)

(hq-util/write-and-cap!
 hq-config people-in-queue
 [{: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"}]
 echo)

(hq-util/write-and-cap!
 hq-config colors-in-queue
 [{: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"}]
 echo)

(def catalog
  [{:onyx/name :people-in
    :onyx/ident :hornetq/read-segments
    :onyx/type :input
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name people-in-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

   {:onyx/name :colors-in
    :onyx/ident :hornetq/read-segments
    :onyx/type :input
    :onyx/medium :hornetq
    :onyx/consumption :concurrent
    :hornetq/queue-name colors-in-queue
    :hornetq/host (:host (:non-clustered (:hornetq config)))
    :hornetq/port (:port (:non-clustered (:hornetq config)))
    :onyx/batch-size batch-size}

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

    :onyx/batch-size batch-size}

   {:onyx/name :children-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}

   {:onyx/name :adults-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}

   {:onyx/name :athletes-wa-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}

   {:onyx/name :everyone-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}

   {:onyx/name :red-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}

   {:onyx/name :blue-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}

   {:onyx/name :green-out
    :onyx/ident :core.async/write-to-chan
    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size batch-size
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}])

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

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

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

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

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

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

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

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

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

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