 (defrecord Message [buf start end])
 
 (defn write-from-buffer [pending-ch publication send-idle-strategy]
-  (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))))
+  (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

(ns onyx.messaging.aeron.publication-manager
  (:require [taoensso.timbre :refer [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]


  (write [this buf start end]
    ;; should possibly use core-async offer here and print out message if going to block
    (when-not (>!! pending-ch (->Message buf start end))))

  (disconnect [this]
    (.close ^Publication @publication)
