<commit_before>(ns ^:no-doc onyx.peer.aggregate
    (:require [clojure.core.async :refer [chan go >! <! <!! >!! close!]]
              [onyx.peer.task-lifecycle-extensions :as l-ext]
              [onyx.peer.pipeline-extensions :as p-ext]
              [onyx.peer.operation :as operation]
              [onyx.extensions :as extensions]
              [onyx.queue.hornetq :refer [take-segments]]
              [onyx.peer.transform :as transformer]
              [taoensso.timbre :refer [debug fatal]]
              [dire.core :refer [with-post-hook!]]))

(defn consumer-loop [event session input consumer halting-ch session-ch]
  (go
   (try
     (loop []
       (let [f (fn [] {:input input
                      :message (extensions/consume-message (:onyx.core/queue event) consumer)})
             msgs (doall (take-segments f (:onyx/batch-size (:onyx.core/task-map event))))]
         (>! session-ch {:session session :halting-ch halting-ch :msgs msgs}))
       (when (<! halting-ch)
         (recur)))
     (catch Exception e
       (fatal e)))))

(defn inject-pipeline-resource-shim
  [{:keys [onyx.core/queue onyx.core/ingress-queues onyx.core/task-map] :as event}]
  (let [session (extensions/bind-active-session queue (first (vals ingress-queues)))
        consumers (map (fn [[task queue-name]]
                         {:input task
                          :consumer (extensions/create-consumer queue session queue-name)})
                       ingress-queues)
        halting-ch (chan 0)
        read-ch (chan 1)
        rets
        {:onyx.aggregate/queue
         {:session session
          :consumers consumers
          :halting-ch halting-ch}
         :onyx.aggregate/read-ch read-ch
         :onyx.core/reserve? true
         :onyx.transform/fn (operation/resolve-fn task-map)}]
    (doseq [c consumers]
      (consumer-loop event session (:input c) (:consumer c) halting-ch read-ch))
    (merge event rets)))

(defn inject-temporal-resource-shim
  [event]
  ;;; To make HornetQ clustered grouping work for Onyx's semantics,
  ;;; only one session can be alive. Open a fake and redefine it later
  ;;; in the pipeline.
  {:onyx.core/session :placeholder})

(defn read-batch-shim [{:keys [onyx.core/queue] :as event}]
  (let [{:keys [session halting-ch msgs]} (<!! (:onyx.aggregate/read-ch event))]
    (doseq [msg msgs]
      (extensions/ack-message queue (:message msg)))
    (merge event<commit_msg>Don't retry cached exhausted chs.<commit_after>(ns ^:no-doc onyx.peer.aggregate
    (:require [clojure.core.async :refer [chan go >! <! <!! >!! close!]]
              [onyx.peer.task-lifecycle-extensions :as l-ext]
              [onyx.peer.pipeline-extensions :as p-ext]
              [onyx.peer.operation :as operation]
              [onyx.extensions :as extensions]
              [onyx.queue.hornetq :refer [take-segments]]
              [onyx.peer.transform :as transformer]
              [taoensso.timbre :refer [debug fatal]]
              [dire.core :refer [with-post-hook!]]))

(defn consumer-loop [event session input consumer halting-ch session-ch]
  (go
   (try
     (loop []
       (let [f (fn [] {:input input
                      :message (extensions/consume-message (:onyx.core/queue event) consumer)})
             msgs (doall (take-segments f (:onyx/batch-size (:onyx.core/task-map event))))]
         (>! session-ch {:session session :halting-ch halting-ch :msgs msgs}))
       (when (<! halting-ch)
         (recur)))
     (catch Exception e
       (fatal e)))))

(defn inject-pipeline-resource-shim
  [{:keys [onyx.core/queue onyx.core/ingress-queues onyx.core/task-map] :as event}]
  (let [session (extensions/bind-active-session queue (first (vals ingress-queues)))
        uncached (into {} (remove (fn [[t _]] (some #{t} learned)) ingress-queues))
        consumers (map (fn [[task queue-name]]
                         {:input task
                          :consumer (extensions/create-consumer queue session queue-name)})
                       uncached)
        halting-ch (chan 0)
        read-ch (chan 1)
        rets
        {:onyx.aggregate/queue
         {:session session
          :consumers consumers
          :halting-ch halting-ch}
         :onyx.aggregate/read-ch read-ch
         :onyx.core/reserve? true
         :onyx.transform/fn (operation/resolve-fn task-map)}]
    (doseq [c consumers]
      (consumer-loop event session (:input c) (:consumer c) halting-ch read-ch))
    (merge event rets)))

(defn inject-temporal-resource-shim
  [event]
  ;;; To make HornetQ clustered grouping work for Onyx's semantics,
  ;;; only one session can be alive. Open a fake and redefine it later
  ;;; in the pipeline.
  {:onyx.core/session :placeholder})

(defn read-batch-shim [{:keys [onyx.core/queue] :as event}]
  (let [{:keys [session halting-ch msgs]} (<!! (:onyx.aggregate/read-ch event))]
    (doseq [msg msgs]
      (extensions/ack-message queue (:message msg)))
    (merge event
{'fraction_non_alphanumeric': 0.14034021871202917, 'fraction_numerical': 0.0008100445524503848, 'mean_word_length': 2.339418526031102, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 9, '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': '4093139', 'n_tokens_mistral': 1629, 'n_tokens_neox': 1551, 'n_words': 398}