   {})
 
 (defmethod p-ext/read-batch :core.async/read-from-chan
-  [{:keys [onyx.core/task-map core.async/chan core.async/retry-ch
-           core.async/pending-messages] :as event}]
+  [{:keys [onyx.core/task-map core.async/chan core.async/retry-ch 
+           core.async/pending-messages core.async/drained?] :as event}]
   (let [pending (count @pending-messages)
         max-pending (or (:onyx/max-pending task-map) (:onyx/max-pending defaults))
         batch-size (:onyx/batch-size task-map)

        timeout-ch (timeout ms)
        batch (if (zero? max-segments)
                (<!! timeout-ch)
                (->> (range max-segments)
                     (map (fn [_]
                            {:id (java.util.UUID/randomUUID)
                             :input :core.async
                             :message (first (alts!! [retry-ch chan timeout-ch] :priority true))}))
                     (remove (comp nil? :message))))]
    (doseq [m batch]
      (swap! pending-messages assoc (:id m) (:message m)))
    {:onyx.core/batch batch}))

(defmethod p-ext/ack-message :core.async/read-from-chan

  (get @pending-messages message-id))

(defmethod p-ext/drained? :core.async/read-from-chan
  [{:keys [core.async/drained? core.async/pending-messages] :as event}]
  @drained?)

(defmethod p-ext/write-batch :core.async/write-to-chan
  [{:keys [onyx.core/results core.async/chan] :as event}]
