                                                    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)))))))

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

    (let [inbound-ch (:inbound-ch (:messenger-buffer component))
          daemon (:acking-daemon component)
          release-ch (chan (clojure.core.async/dropping-buffer 100000))

          ;; TODO, correct port allocation
          port (+ 40000 (rand-int 10000))
          channel (format "udp://%s:%s" (choose-ip opts) port)
          ip (choose-ip opts)

          rand-stream-id (fn [] (rand-int 1000000))
          send-stream-id (rand-stream-id)
          acker-stream-id (rand-stream-id)
          completion-stream-id (rand-stream-id)

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

          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))))]
      (assoc component 
             :channel channel 
             :ip ip
             :send-stream-id send-stream-id
             :acker-stream-id acker-stream-id 
             :completion-stream-id completion-stream-id
             :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
             :release-ch release-ch 
             :context ctx 
             :aeron aeron)))

  (stop [{:keys [aeron send-subscriber acker-subscriber 
                 completion-subscriber accept-send-fut accept-acker-fut
                 accept-completion-fut release-ch] :as component}]
    (taoensso.timbre/info "Stopping Aeron")

    (try 
      (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 aeron (.close aeron))
      (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 :driver nil :ip nil :channel nil :release-ch nil :aeron nil 
           :accept-send-fut nil :accept-acker-fut nil :accept-completion-fut nil 
           :send-subscriber nil :acker-subscriber nil :completion-subscriber nil)))

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

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

(defmethod extensions/send-peer-site AeronConnection
  [messenger]
  {:channel (:channel messenger)
   :stream-id (:send-stream-id messenger)})

(defmethod extensions/acker-peer-site AeronConnection
  [messenger]
  {:channel (:channel messenger)
   :stream-id (:acker-stream-id messenger)})

(defmethod extensions/completion-peer-site AeronConnection
  [messenger]
  {:channel (:channel messenger)
   :stream-id (:completion-stream-id messenger)})

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

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

(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 (:send-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 (:acker-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 (:completion-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 (:send-pub peer-link))
  (.close (:acker-pub peer-link))
  (.close (:completion-pub peer-link))
  (.close (:conn peer-link)) 
  {})
