<commit_before>(ns ^:no-doc onyx.peer.virtual-peer
  (:require [clojure.core.async :refer [chan alts!! >!! <!! close!]]
            [com.stuartsierra.component :as component]
            [taoensso.timbre :as timbre]
            [dire.core :as dire]
            [onyx.extensions :as extensions]
            [onyx.peer.task-pipeline :refer [task-pipeline]]))

(defn payload-loop [id sync queue payload-ch shutdown-ch status-ch dead-ch pulse fn-params]
  (let [complete-ch (chan 1)]
    (loop [pipeline nil]
      (when-let [[v ch] (alts!! [shutdown-ch complete-ch payload-ch] :priority true)]
        (when-not (nil? pipeline)
          (component/stop pipeline))

        (cond (nil? v) (extensions/delete sync pulse)
              (= ch complete-ch) (recur nil)
              (= ch payload-ch)
              (let [payload-node (:path v)
                    payload (extensions/read-place sync payload-node)
                    status-ch (chan 1)]

                (extensions/on-change sync (:status (:nodes payload)) #(>!! status-ch %))
                (extensions/touch-place sync (:ack (:nodes payload)))

                (<!! status-ch)

                (let [new-pipeline (task-pipeline id payload sync queue payload-ch complete-ch fn-params)]
                  (recur (component/start new-pipeline)))))))
    (>!! dead-ch true)))

(defrecord VirtualPeer [fn-params]
  component/Lifecycle

  (start [{:keys [sync queue] :as component}]<commit_msg>Adds else to cond.<commit_after>(ns ^:no-doc onyx.peer.virtual-peer
  (:require [clojure.core.async :refer [chan alts!! >!! <!! close!]]
            [com.stuartsierra.component :as component]
            [taoensso.timbre :as timbre]
            [dire.core :as dire]
            [onyx.extensions :as extensions]
            [onyx.peer.task-pipeline :refer [task-pipeline]]))

(defn payload-loop [id sync queue payload-ch shutdown-ch status-ch dead-ch pulse fn-params]
  (let [complete-ch (chan 1)]
    (loop [pipeline nil]
      (when-let [[v ch] (alts!! [shutdown-ch complete-ch payload-ch] :priority true)]
        (when-not (nil? pipeline)
          (component/stop pipeline))

        (cond (nil? v) (extensions/delete sync pulse)
              (= ch complete-ch) (recur nil)
              (= ch payload-ch)
              (let [payload-node (:path v)
                    payload (extensions/read-place sync payload-node)
                    status-ch (chan 1)]

                (extensions/on-change sync (:status (:nodes payload)) #(>!! status-ch %))
                (extensions/touch-place sync (:ack (:nodes payload)))

                (<!! status-ch)

                (let [new-pipeline (task-pipeline id payload sync queue payload-ch complete-ch fn-params)]
                  (recur (component/start new-pipeline))))
              :else (recur nil))))
    (>!! dead-ch true)))

(defrecord VirtualPeer [fn-params]
  component/Lifecycle

  (start [{:keys [sync queue] :as component}]
{'fraction_non_alphanumeric': 0.14280856851411083, 'fraction_numerical': 0.0013600816048962938, 'mean_word_length': 2.163440860215054, 'pattern_counts': {'":': 0, '<': 7, '<?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': '5300525', 'n_tokens_mistral': 899, 'n_tokens_neox': 870, 'n_words': 237}