                                                    10
                                                    (.toNanos TimeUnit/MICROSECONDS 1)
                                                    (.toNanos TimeUnit/MICROSECONDS 100)))]
+        (timbre/info "Started reading")
         (while (not (Thread/interrupted))
           (let [fragments-read (.poll ^uk.co.real_logic.aeron.Subscription subscription limit)]
             (.idle idle-strategy fragments-read)))))))

(defmethod extensions/send-messages AeronConnection
  [messenger event peer-link batch]
  (let [[len unsafe-buffer] (protocol/build-messages-msg-buf batch)
        pub ^uk.co.real_logic.aeron.Publication (:pub peer-link)
        offer-f (fn [] (.offer pub unsafe-buffer 0 len))]
    (while (not (offer-f)))))

(defmethod extensions/internal-ack-message AeronConnection
  [messenger event peer-link message-id completion-id ack-val]
  (let [unsafe-buffer (protocol/build-acker-message message-id completion-id ack-val)
        pub ^uk.co.real_logic.aeron.Publication (:pub peer-link)
        offer-f (fn [] (.offer pub unsafe-buffer 0 protocol/ack-msg-length))]
    (while (not (offer-f)))))

(defmethod extensions/internal-complete-message AeronConnection
  [messenger event id peer-link]
  (let [unsafe-buffer (protocol/build-completion-msg-buf id)
        pub ^uk.co.real_logic.aeron.Publication (:pub peer-link) 
        offer-f (fn [] (.offer pub unsafe-buffer 0 protocol/completion-msg-length))]
    (while (not (offer-f)))))

(defmethod extensions/close-peer-connection AeronConnection
  [messenger event peer-link]
  (.close (:pub peer-link))
  (.close (:conn peer-link)) {})

  (start [component]
    (taoensso.timbre/info "Starting Aeron")

    (let [release-ch (chan (clojure.core.async/dropping-buffer 100000))
          ip (choose-ip opts)]
      (assoc component 
             :ip ip
             :site-resources (atom nil)
             :release-ch release-ch)))

  (stop [{:keys [aeron site-resources release-ch] :as component}]
    (taoensso.timbre/info "Stopping Aeron")
    (try 
      (when-let [resources @site-resources]
        (let [{:keys [conn
                      send-subscriber 
                      acker-subscriber 
                      completion-subscriber 
                      accept-send-fut 
                      accept-acker-fut
                      accept-completion-fut]} resources] 
          (future-cancel accept-send-fut)
          (future-cancel accept-acker-fut)
          (future-cancel accept-completion-fut)
          (when send-subscriber (.close send-subscriber))
          (when acker-subscriber (.close acker-subscriber))
          (when completion-subscriber (.close completion-subscriber))
          (when conn (.close conn)))
        (reset! site-resources nil))
      (close! (:release-ch component))

      ;; FIXME, need to startup and shutdown aeron resources properly
      ;; however, this can't be done per connection
      ;(CloseHelper/quietClose (:driver component))

      (catch Exception e
        (fatal e)))

    (assoc component 
           :ip nil 
           :site-resources nil
           :release-ch nil)))

(defn aeron [opts]
  (map->AeronConnection {:opts opts}))

(defmethod extensions/peer-site AeronConnection
  [messenger]
  {:aeron/ip (:ip messenger)})

(def allc (atom 40200))

(defmethod extensions/assign-site-resources AeronConnection
  [messenger peer-sites]
  {:aeron/port (swap! allc inc)})

(def send-stream-id 1)
(def acker-stream-id 2)
(def completion-stream-id 3)

(defn aeron-channel [ip port]
  (format "udp://%s:%s" ip port))

(defmethod extensions/open-peer-site AeronConnection
  [messenger assigned]
  (timbre/info "Open peer site " messenger assigned)
  (let [inbound-ch (:inbound-ch (:messenger-buffer messenger))
        release-ch (:release-ch messenger)
        daemon (:acking-daemon messenger)

        ctx (.errorHandler (Aeron$Context.) no-op-error-handler)
        aeron (Aeron/connect ctx)

        channel (aeron-channel (:ip messenger) (:aeron/port assigned))

        send-handler (data-handler (partial handle-sent-message inbound-ch))
        acker-handler (data-handler (partial handle-acker-message daemon))
        completion-handler (data-handler (partial handle-completion-message release-ch))

        send-subscriber (.addSubscription aeron channel send-stream-id send-handler)
        acker-subscriber (.addSubscription aeron channel acker-stream-id acker-handler)
        completion-subscriber (.addSubscription aeron channel completion-stream-id completion-handler)
        accept-send-fut (future (try (.accept ^Consumer (consumer 10) send-subscriber) 
                                     (catch Exception e (fatal e))))
        accept-acker-fut (future (try (.accept ^Consumer (consumer 10) acker-subscriber) 
                                      (catch Exception e (fatal e))))
        accept-completion-fut (future (try (.accept ^Consumer (consumer 10) completion-subscriber) 
                                           (catch Exception e (fatal e))))]
    (reset! (:site-resources messenger)
            {:conn aeron
             :accept-send-fut accept-send-fut
             :accept-acker-fut accept-acker-fut
             :accept-completion-fut accept-completion-fut
             :send-subscriber send-subscriber
             :acker-subscriber acker-subscriber
             :completion-subscriber completion-subscriber})))

(defmethod extensions/connect-to-peer AeronConnection
  [messenger event {:keys [aeron/ip]} {:keys [aeron/port]}]
  (timbre/info "Connect to peer " ip port)
  (let [ctx (.errorHandler (Aeron$Context.) no-op-error-handler)
        aeron (Aeron/connect ctx)
        channel (aeron-channel ip port)
        send-pub (.addPublication aeron channel send-stream-id)
        acker-pub (.addPublication aeron channel acker-stream-id)
        completion-pub (.addPublication aeron channel completion-stream-id)]
    {:conn aeron :send-pub send-pub :acker-pub acker-pub :completion-pub completion-pub}))

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