         (persistent! coll)
         (recur (inc n) 
                (conj! coll (.nextLong rng)))))))
-
-

    (:require [clojure.core.async :refer [chan >!! close! sliding-buffer]]
              [com.stuartsierra.component :as component]
              [onyx.static.default-vals :refer [defaults]]
              [onyx.types :refer [->Ack]]
              [taoensso.timbre :as timbre]))

(defrecord AckingDaemon [opts]

(defn ack-message [daemon message-id completion-id ack-val]
  (let [rets
        (swap!
         (:ack-state daemon)
         (fn [state]
           (if-let [current-ack-val (get-in state [message-id :ack-val])]
             (assoc-in state [message-id :ack-val] (bit-xor current-ack-val ack-val))
             (assoc state message-id {:completion-id completion-id
                                      :ack-val ack-val}))))]

    (when-let [x (get rets message-id)]
      (when (zero? (:ack-val x))
        (swap! (:ack-state daemon) dissoc message-id)
        (>!! (:completions-ch daemon) {:id message-id
                                       :peer-id completion-id})))))

(defn gen-message-id
  "Generates a unique ID for a message - acts as the root id."
