<commit_before>(ns ^:no-doc onyx.messaging.aeron.publication-manager
  (:require [taoensso.timbre :refer [error fatal info] :as timbre]
            [clojure.core.async :refer [chan >!! <!! put! close! sliding-buffer thread]])
  (:import [uk.co.real_logic.aeron Aeron Aeron$Context FragmentAssembler Publication Subscription AvailableImageHandler]
           [uk.co.real_logic.agrona ErrorHandler CloseHelper]
           [uk.co.real_logic.aeron.exceptions DriverTimeoutException]
           [uk.co.real_logic.agrona.concurrent 
            UnsafeBuffer IdleStrategy BackoffIdleStrategy BusySpinIdleStrategy]))

(defprotocol PPublicationManager
  (write [this buf start end])
  (connect [_])
  (start [_])
  (stop [_]))

(defrecord Message [buf start end])

(defn write-from-buffer [publication-manager pending-ch send-idle-strategy]
  (try 
    (let [pub (:publication publication-manager)] 
      (loop [] 
        (when-let [msg (<!! pending-ch)]
          (while (let [result ^long (.offer ^Publication pub (:buf msg) (:start msg) (:end msg))]
                   (or (= result Publication/BACK_PRESSURED)
                       (= result Publication/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 DriverTimeoutException e
      (fatal "Aeron write failure due to driver time out." e)
      (stop publication-manager))
    (catch IllegalStateException e
      (fatal "Writing to closed publication." e)
      (stop publication-manager))
    (catch Throwable t
      (fatal "Aeron write from buffer error: " t)
      (stop publication-manager))))

(defrecord PublicationManager [channel stream-id send-idle-strategy connection publication pending-ch write-fut cleanup-fn]
  PPublicationManager
  (start [this]
    (assoc this :write-fut (future (write-from-buffer this pending-ch send-idle-strategy))))

  (write [this buf start end]
    ;; use core.async offer! here and call monitoring when blocked
    ;; will only be possible when core.async releases master again
    (>!! pending-ch (->Message buf start end)))

  (stop [this]
    (info (format "Stopped publication manager at: %s, stream-id: %s" channel stream-id))
    (cleanup-fn)
    (future-cancel (:write-fut this))
    (close! pending-ch)
    (try (.close ^Publication publication)
         (catch Throwable t 
           (info "Could not close publication:" t)))
    (try (.close ^Aeron connection)
         (catch Throwable t 
           (info "Could not close connection" t)))
    this)

  (connect [this]
    (let [error-handler (reify ErrorHandler
                          (onError [this x] 
                            (taoensso.timbre/warn "Aeron messaging publication error:" x)))
          conn (Aeron/connect (.errorHandler (Aeron$Context.) error-handler))
          pub (.addPublication conn channel stream-id)]
      (info (format "Connected via publication on: %s, stream-id: %s" channel stream-id))
      (assoc this :publication pub :connection conn))))

(defn new-publication-manager [channel stream-id send-idle-strategy write-buffer-size cleanup-fn]
  (->PublicationManager channel stream-id send-idle-strategy 
                        nil nil (chan write-buffer-size) nil cleanup-fn))

<commit_msg>Improve log message, created pub, not connectd<commit_after>(ns ^:no-doc onyx.messaging.aeron.publication-manager
  (:require [taoensso.timbre :refer [error fatal info] :as timbre]
            [clojure.core.async :refer [chan >!! <!! put! close! sliding-buffer thread]])
  (:import [uk.co.real_logic.aeron Aeron Aeron$Context FragmentAssembler Publication Subscription AvailableImageHandler]
           [uk.co.real_logic.agrona ErrorHandler CloseHelper]
           [uk.co.real_logic.aeron.exceptions DriverTimeoutException]
           [uk.co.real_logic.agrona.concurrent 
            UnsafeBuffer IdleStrategy BackoffIdleStrategy BusySpinIdleStrategy]))

(defprotocol PPublicationManager
  (write [this buf start end])
  (connect [_])
  (start [_])
  (stop [_]))

(defrecord Message [buf start end])

(defn write-from-buffer [publication-manager pending-ch send-idle-strategy]
  (try 
    (let [pub (:publication publication-manager)] 
      (loop [] 
        (when-let [msg (<!! pending-ch)]
          (while (let [result ^long (.offer ^Publication pub (:buf msg) (:start msg) (:end msg))]
                   (or (= result Publication/BACK_PRESSURED)
                       (= result Publication/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 DriverTimeoutException e
      (fatal "Aeron write failure due to driver time out." e)
      (stop publication-manager))
    (catch IllegalStateException e
      (fatal "Writing to closed publication." e)
      (stop publication-manager))
    (catch Throwable t
      (fatal "Aeron write from buffer error: " t)
      (stop publication-manager))))

(defrecord PublicationManager [channel stream-id send-idle-strategy connection publication pending-ch write-fut cleanup-fn]
  PPublicationManager
  (start [this]
    (assoc this :write-fut (future (write-from-buffer this pending-ch send-idle-strategy))))

  (write [this buf start end]
    ;; use core.async offer! here and call monitoring when blocked
    ;; will only be possible when core.async releases master again
    (>!! pending-ch (->Message buf start end)))

  (stop [this]
    (info (format "Stopped publication manager at: %s, stream-id: %s" channel stream-id))
    (cleanup-fn)
    (future-cancel (:write-fut this))
    (close! pending-ch)
    (try (.close ^Publication publication)
         (catch Throwable t 
           (info "Could not close publication:" t)))
    (try (.close ^Aeron connection)
         (catch Throwable t 
           (info "Could not close connection" t)))
    this)

  (connect [this]
    (let [error-handler (reify ErrorHandler
                          (onError [this x] 
                            (taoensso.timbre/warn "Aeron messaging publication error:" x)))
          conn (Aeron/connect (.errorHandler (Aeron$Context.) error-handler))
          pub (.addPublication conn channel stream-id)]
      (info (format "Created publication at: %s, stream-id: %s" channel stream-id))
      (assoc this :publication pub :connection conn))))

(defn new-publication-manager [channel stream-id send-idle-strategy write-buffer-size cleanup-fn]
  (->PublicationManager channel stream-id send-idle-strategy 
                        nil nil (chan write-buffer-size) nil cleanup-fn))


{'fraction_non_alphanumeric': 0.11152196420647834, 'fraction_numerical': 0.00029581422866439876, 'mean_word_length': 2.9359720605355064, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3011217', 'n_tokens_mistral': 1998, 'n_tokens_neox': 1909, 'n_words': 595}