<commit_before>(ns ^:no-doc onyx.messaging.aeron
    (:require [clojure.core.async :refer [chan >!! <!! alts!! timeout close! dropping-buffer]]
              [com.stuartsierra.component :as component]
              [taoensso.timbre :refer [fatal] :as timbre]
              [onyx.messaging.protocol-aeron :as protocol]
              [onyx.messaging.acking-daemon :as acker]
              [onyx.messaging.common :refer [bind-addr external-addr allowable-ports]]
              [onyx.extensions :as extensions]
              [onyx.compression.nippy :refer [compress decompress]])
    (:import [uk.co.real_logic.aeron Aeron FragmentAssemblyAdapter]
             [uk.co.real_logic.aeron Aeron$Context]
             [uk.co.real_logic.aeron.driver MediaDriver MediaDriver$Context ThreadingMode]
             [uk.co.real_logic.aeron.common.concurrent.logbuffer DataHandler]
             [uk.co.real_logic.agrona.concurrent UnsafeBuffer]
             [uk.co.real_logic.agrona CloseHelper]
             [uk.co.real_logic.agrona.concurrent IdleStrategy BackoffIdleStrategy]
             [java.util.function Consumer]
             [java.util.concurrent TimeUnit]))

(defrecord AeronPeerGroup [opts]
  component/Lifecycle
  (start [component]
    (taoensso.timbre/info "Starting Aeron Peer Group")
    (if (and (contains? opts :onyx.messaging.aeron/embedded-driver)
             (false? (opts :onyx.messaging.aeron/embedded-driver)))
      component
      (let [ctx (doto (MediaDriver$Context.) 
                  (.threadingMode ThreadingMode/SHARED)
                  (.dirsDeleteOnExit true))
            media-driver (MediaDriver/launch ctx)]
        (assoc component :media-driver media-driver))))

  (stop [{:keys [media-driver] :as component}]
    (taoensso.timbre/info "Stopping Aeron Peer Group")
    (when media-driver (.close media-driver))
    (assoc component :media-driver nil)))

(defn aeron-peer-group [opts]
  (map->AeronPeerGroup {:opts opts}))

(defmethod extensions/assign-site-resources :aeron
  [config peer-site peer-sites]
  (let [used-ports (->> (vals peer-sites) <commit_msg>Switch back to dedicated mode now that we have less subscriptions.<commit_after>(ns ^:no-doc onyx.messaging.aeron
    (:require [clojure.core.async :refer [chan >!! <!! alts!! timeout close! dropping-buffer]]
              [com.stuartsierra.component :as component]
              [taoensso.timbre :refer [fatal] :as timbre]
              [onyx.messaging.protocol-aeron :as protocol]
              [onyx.messaging.acking-daemon :as acker]
              [onyx.messaging.common :refer [bind-addr external-addr allowable-ports]]
              [onyx.extensions :as extensions]
              [onyx.compression.nippy :refer [compress decompress]])
    (:import [uk.co.real_logic.aeron Aeron FragmentAssemblyAdapter]
             [uk.co.real_logic.aeron Aeron$Context]
             [uk.co.real_logic.aeron.driver MediaDriver MediaDriver$Context ThreadingMode]
             [uk.co.real_logic.aeron.common.concurrent.logbuffer DataHandler]
             [uk.co.real_logic.agrona.concurrent UnsafeBuffer]
             [uk.co.real_logic.agrona CloseHelper]
             [uk.co.real_logic.agrona.concurrent IdleStrategy BackoffIdleStrategy]
             [java.util.function Consumer]
             [java.util.concurrent TimeUnit]))

(defrecord AeronPeerGroup [opts]
  component/Lifecycle
  (start [component]
    (taoensso.timbre/info "Starting Aeron Peer Group")
    (if (and (contains? opts :onyx.messaging.aeron/embedded-driver)
             (false? (opts :onyx.messaging.aeron/embedded-driver)))
      component
      (let [ctx (doto (MediaDriver$Context.) 
                  (.threadingMode ThreadingMode/DEDICATED)
                  (.dirsDeleteOnExit true))
            media-driver (MediaDriver/launch ctx)]
        (assoc component :media-driver media-driver))))

  (stop [{:keys [media-driver] :as component}]
    (taoensso.timbre/info "Stopping Aeron Peer Group")
    (when media-driver (.close media-driver))
    (assoc component :media-driver nil)))

(defn aeron-peer-group [opts]
  (map->AeronPeerGroup {:opts opts}))

(defmethod extensions/assign-site-resources :aeron
  [config peer-site peer-sites]
  (let [used-ports (->> (vals peer-sites) 
{'fraction_non_alphanumeric': 0.11938534278959811, 'fraction_numerical': 0.0, 'mean_word_length': 2.7980251346499103, 'pattern_counts': {'":': 0, '<': 5, '<?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': '6932689', 'n_tokens_mistral': 1351, 'n_tokens_neox': 1289, 'n_words': 268}