 
 (def config (read-string (slurp (clojure.java.io/resource "test-config.edn"))))
 
-(def scheduler :onyx.job-scheduler/percentage)
-
 (def env-config
-  {:hornetq/mode :udp
-   :hornetq/server? true
-   :hornetq.server/type :embedded
-   :hornetq.udp/cluster-name (:cluster-name (:hornetq config))
-   :hornetq.udp/group-address (:group-address (:hornetq config))
-   :hornetq.udp/group-port (:group-port (:hornetq config))
-   :hornetq.udp/refresh-timeout (:refresh-timeout (:hornetq config))
-   :hornetq.udp/discovery-timeout (:discovery-timeout (:hornetq config))
-   :hornetq.embedded/config (:configs (:hornetq config))
-   :zookeeper/address (:address (:zookeeper config))
+  {:zookeeper/address (:address (:zookeeper config))
    :zookeeper/server? true
    :zookeeper.server/port (:spawn-port (:zookeeper config))
-   :onyx.peer/job-scheduler scheduler
    :onyx/id onyx-id})
 
 (def peer-config
-  {:hornetq/mode :udp
-   :hornetq.udp/cluster-name (:cluster-name (:hornetq config))
-   :hornetq.udp/group-address (:group-address (:hornetq config))
-   :hornetq.udp/group-port (:group-port (:hornetq config))
-   :hornetq.udp/refresh-timeout (:refresh-timeout (:hornetq config))
-   :hornetq.udp/discovery-timeout (:discovery-timeout (:hornetq config))
-   :zookeeper/address (:address (:zookeeper config))
+  {:zookeeper/address (:address (:zookeeper config))
    :onyx/id onyx-id
-   :onyx.peer/inbox-capacity (:inbox-capacity (:peer config))
-   :onyx.peer/outbox-capacity (:outbox-capacity (:peer config))
-   :onyx.peer/job-scheduler scheduler})
+   :onyx.peer/job-scheduler :onyx.job-scheduler/percentage
+   :onyx.messaging/impl :http-kit-websockets})
 
 (def env (onyx.api/start-env env-config))
 
 (def catalog-1
   [{:onyx/name :a
-    :onyx/ident :hornetq/read-segments
+    :onyx/ident :core.async/read-from-chan
     :onyx/type :input
-    :onyx/medium :hornetq
-    :onyx/batch-size 20}
+    :onyx/medium :core.async
+    :onyx/batch-size 20
+    :onyx/doc "Reads segments from a core.async channel"}
 
    {:onyx/name :b
-    :onyx/ident :hornetq/write-segments
+    :onyx/ident :core.async/write-to-chan
     :onyx/type :output
-    :onyx/medium :hornetq
-    :onyx/batch-size 20}])
+    :onyx/medium :core.async
+    :onyx/batch-size 20
+    :onyx/doc "Writes segments to a core.async channel"}])
 
 (def catalog-2
   [{:onyx/name :c
-    :onyx/ident :hornetq/read-segments
+    :onyx/ident :core.async/read-from-chan
     :onyx/type :input
-    :onyx/medium :hornetq
-    :onyx/batch-size 20}
+    :onyx/medium :core.async
+    :onyx/batch-size 20
+    :onyx/doc "Reads segments from a core.async channel"}
 
    {:onyx/name :d
-    :onyx/ident :hornetq/write-segments
+    :onyx/ident :core.async/write-to-chan
     :onyx/type :output
-    :onyx/medium :hornetq
-    :onyx/batch-size 20}])
+    :onyx/medium :core.async
+    :onyx/batch-size 20
+    :onyx/doc "Writes segments to a core.async channel"}])
+
+(def a-chan (chan 100))
+
+(def b-chan (chan (sliding-buffer 100)))
+
+(def c-chan (chan 100))
+
+(def d-chan (chan (sliding-buffer 100)))
+
+(defmethod l-ext/inject-lifecycle-resources :a
+  [_ _] {:core.async/in-chan a-chan})
+
+(defmethod l-ext/inject-lifecycle-resources :b
+  [_ _] {:core.async/out-chan b-chan})
+
+(defmethod l-ext/inject-lifecycle-resources :c
+  [_ _] {:core.async/in-chan c-chan})
+
+(defmethod l-ext/inject-lifecycle-resources :d
+  [_ _] {:core.async/out-chan d-chan})
 
 (def j1
-  (onyx.api/submit-job peer-config
-                       {:workflow [[:a :b]]
-                        :catalog catalog-1
-                        :percentage 70
-                        :task-scheduler :onyx.task-scheduler/round-robin}))
+  (onyx.api/submit-job
+   peer-config
+   {:workflow [[:a :b]]
+    :catalog catalog-1
+    :percentage 70
+    :task-scheduler :onyx.task-scheduler/round-robin}))
 
 (def j2
-  (onyx.api/submit-job peer-config
-                       {:workflow [[:c :d]]
-                        :catalog catalog-2
-                        :percentage 30
-                        :task-scheduler :onyx.task-scheduler/round-robin}))
+  (onyx.api/submit-job
+   peer-config
+   {:workflow [[:c :d]]
+    :catalog catalog-2
+    :percentage 30
+    :task-scheduler :onyx.task-scheduler/round-robin}))
 
 (def n-peers 10)
 
-(def v-peers-1 (onyx.api/start-peers! n-peers peer-config))
+(def v-peers-1 (onyx.api/start-peers n-peers peer-config))
 
 (def ch (chan n-peers))
 


(fact "70/30% split for percentage job scheduler succeeded" true => true)

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

(def ch (chan n-peers))


(ns onyx.log.percentage-multi-job-test
  (:require [clojure.core.async :refer [chan >!! <!! close! sliding-buffer]]
            [onyx.extensions :as extensions]
            [onyx.peer.task-lifecycle-extensions :as l-ext]
            [onyx.plugin.core-async :refer [take-segments!]]
            [onyx.api :as api]
            [midje.sweet :refer :all]
            [zookeeper :as zk]))
