 (ns ^:no-doc onyx.messaging.aeron
     (:require [clojure.core.async :refer [chan >!! <!! alts!! timeout close!]]
               [com.stuartsierra.component :as component]
-              [taoensso.timbre :refer [warn] :as timbre]
+              [taoensso.timbre :refer [fatal] :as timbre]
               [onyx.messaging.acking-daemon :as acker]
               [onyx.compression.nippy :refer [compress decompress]]
               [onyx.extensions :as extensions])

  [messenger event {:keys [channel stream-id]}]
  (let [ctx (.errorHandler (Aeron$Context.) no-op-error-handler)
        aeron (Aeron/connect ctx)]
    (.addPublication aeron channel stream-id)))

(defmethod extensions/receive-messages AeronConnection
  [messenger {:keys [onyx.core/task-map] :as event}]

  (let [compressed (compress {:id message-id :completion-id completion-id :ack-val ack-val})
        len (count compressed)
        unsafe-buffer (UnsafeBuffer. compressed)
        offer-f (fn [] (.offer (:pub peer-link) unsafe-buffer 0 len))]
    (while (not (offer-f)))))

(defmethod extensions/internal-complete-message AeronConnection
