   (:require [onyx.log.curator :as curator]
             [taoensso.timbre :refer [info error warn trace fatal] :as timbre]
             [com.stuartsierra.component :as component]
-            [clojure.core.async :refer  [chan timeout thread go >! <! <!! >!! alts!! close!]]
+            [clojure.core.async :refer [chan timeout thread go >! <! <!! >!! alts!! close! poll!]]
+            [clojure.core.async.impl.protocols :refer [closed?]]
             [onyx.compression.nippy :as nippy]
             [onyx.extensions :as extensions]
             [onyx.monitoring.measurements :refer [emit-latency-value emit-latency]]

                 :task-id task-id
                 :slot-id slot-id
                 :ledger-id new-ledger-id}})
    (while (and (first (alts!! [kill-ch task-kill-ch] :default true))
                (not= new-ledger-id (last (event->ledger-ids event))))
      (info "New ledger id has not been published yet. Backing off.")
      (Thread/sleep (arg-or-default :onyx.bookkeeper/ledger-id-written-back-off peer-opts))))) 

(defmethod state-extensions/initialize-log :bookkeeper [log-type {:keys [onyx.core/peer-opts] :as event}] 
  (let [bk-client (bookkeeper peer-opts)

          (.asyncAddEntry compacted-ledger 
                          compacted-serialized
                          HandleWriteCallback
                          (list (fn []
                                  (close-handle previous-handle)
                                  (emit-latency-value :window-log-compaction monitoring (- (System/currentTimeMillis) start-time))
                                  (>!! outbox-ch
                                       {:fn :compact-bookkeeper-log-ids
                                        :args {:job-id job-id
                                               :task-id task-id
                                               :slot-id slot-id
                                               :peer-id id
                                               :prev-ledger-ids (vec (butlast current-ids))
                                               :new-ledger-ids [compacted-ledger-id]}}))
                                (fn [] 
                                  (close! (:onyx.core/restart-ch event))))))))))

(defn take-write-batch [batch-size batch-ch]
  (loop [entries [] ack-fns [] i 0]
    (if (< i batch-size)
      (let [[entry ack-fn] (poll! batch-ch)]
        (if entry
          (recur (conj entries entry)
                 (conj ack-fns ack-fn)
                 (inc i))
          [entries ack-fns]))
      [entries ack-fns])))

(defn process-batches [{:keys [ledger-handle next-ledger-handle batch-ch] :as log}
                       {:keys [onyx.core/kill-ch onyx.core/task-kill-ch onyx.core/peer-opts] :as event}]
  (thread
    (let [batch-size (arg-or-default :onyx.bookkeeper/write-batch-size peer-opts)
          batch-backoff (arg-or-default :onyx.bookkeeper/write-batch-backoff peer-opts)]
      (loop [[batch ack-fns] (take-write-batch batch-size batch-ch)]
        ;; Safe point to transition to the next ledger handle
        (when @next-ledger-handle
          (compaction-transition log event))
        (if (empty? batch)
          (Thread/sleep batch-backoff)
          (.asyncAddEntry ^LedgerHandle @ledger-handle 
                          ^bytes (nippy/window-log-compress batch)
                          HandleWriteCallback
                          (list (fn [] (run! (fn [f] (f)) ack-fns))
                                (fn [] (close! (:onyx.core/restart-ch event))))))
        (if (and (not (closed? kill-ch))
                 (not (closed? task-kill-ch)))
          (recur (take-write-batch batch-size batch-ch)))))
          (info "BookKeeper: shutting down batch processing")))

(defn assign-bookkeeper-log-id-spin [{:keys [onyx.core/peer-opts
                                             onyx.core/job-id onyx.core/task-id
