 (ns onyx.messaging.aeron.publication-manager
-  (:require [taoensso.timbre :refer [fatal info] :as timbre]
+  (:require [taoensso.timbre :refer [error fatal info] :as timbre]
             [clojure.core.async :refer [chan >!! <!! close! sliding-buffer thread]])
   (:import [uk.co.real_logic.aeron Aeron Aeron$Context FragmentAssembler Publication Subscription AvailableImageHandler]
            [uk.co.real_logic.agrona ErrorHandler CloseHelper]

    (let [error-handler (reify ErrorHandler
                          (onError [this x] 
                            (taoensso.timbre/warn "Aeron messaging publication error:" x)
                            (taoensso.timbre/warn "Resetting Aeron publication.")
                            (reset-connection this)))

          conn (Aeron/connect (.errorHandler (Aeron$Context.) error-handler))
          pub (.addPublication conn channel stream-id)]
      (reset! publication pub)

(defrecord Message [buf start end])

(defn write-from-buffer [pending-ch publication send-idle-strategy]
  (try 
    (loop [] 
      (when-let [msg (<!! pending-ch)]
        (while (if-let [pub @publication]
                 (let [result ^long (.offer pub (:buf msg) (:start msg) (:end msg))]
                   (or (= result backpressured)
                       (= result not-connected))))
          ;; idle for different amounts of time depending on whether backpressuring or not?
          (.idle ^IdleStrategy send-idle-strategy 0))
        (recur)))
    (catch InterruptedException e)
    (catch Throwable t
      (error "Aeron write from buffer error: " t))))

(defrecord PublicationManager [send-idle-strategy channel stream-id connection publication pending-ch write-fut]
  PPublicationManager
