 (ns ^:no-doc onyx.messaging.aeron
   (:require [clojure.core.async :refer [chan >!! <!! alts!! timeout close! sliding-buffer]]
             [com.stuartsierra.component :as component]
-            [taoensso.timbre :refer [fatal info] :as timbre]
+            [taoensso.timbre :refer [fatal info warn] :as timbre]
             [onyx.messaging.aeron.peer-manager :as pm]
             [onyx.messaging.aeron.publication-manager :as pubm]
             [onyx.messaging.aeron.publication-pool :as pub-pool :refer [get-publication]]

                                                (.toNanos TimeUnit/MICROSECONDS 10)
                                                (.toNanos TimeUnit/MICROSECONDS 1000))))

(defn consumer [handler ^IdleStrategy idle-strategy limit]
  (reify Consumer
    (accept [this subscription]
      (while (not (Thread/interrupted))
        (let [fragments-read (.poll ^Subscription subscription ^FragmentHandler handler ^int limit)]
          (.idle idle-strategy fragments-read))))))


             :virtual-peers virtual-peers
             :compress-f compress-f
             :decompress-f decompress-f
             :shutdown shutdown
             :port port
             :send-idle-strategy send-idle-strategy
             :subscriber-count subscriber-count
             :subscribers subscribers)))

  (stop [{:keys [media-driver media-driver-context subscribers publication-pool shutdown] :as component}]
    (taoensso.timbre/info "Stopping Aeron Peer Group")
    (component/stop publication-pool)
    (reset! shutdown true)
     
    (when media-driver (.close ^MediaDriver media-driver))
    (when media-driver-context (.deleteAeronDirectory ^MediaDriver$Context media-driver-context))
