 
 
 (defn add-core-async-input
-  ([job task opts] (add-core-async-input job task opts 1000))
+  ([job task opts] (add-core-async-input job task opts default-channel-size))
   ([job task opts chan-size]
    (-> job
        (update :catalog conj (merge {:onyx/name task


(def channels (atom {}))

(defn get-channel
  ([id] (get-channel id nil))
  ([id size]
   (if-let [id (get @channels id)]
     id
     (do (swap! channels assoc id (chan (or size 1000)))
         (get-channel id)))))

(defn inject-in-ch
  [_ lifecycle]
  {:core.async/chan (get-channel (:core.async/id lifecycle)
                                 (or (:core.async/size lifecycle) 1000))})
(defn inject-out-ch
  [_ lifecycle]
  {:core.async/chan (get-channel (:core.async/id lifecycle)
                                 (or (:core.async/size lifecycle) 1000))})

(def in-calls
  {:lifecycle/before-task-start inject-in-ch})

                                  :lifecycle/calls :onyx.plugin.core-async/reader-calls}]))))

(defn add-core-async-output
  ([job task opts] (add-core-async-output job task opts default-channel-size))
  ([job task opts chan-size]
   (-> job
       (update :catalog conj (merge {:onyx/name task
