<commit_before>(ns onyx.flow-conditions.fc-routing
  (:require [onyx.lifecycles.lifecycle-invoke :as lc]
            [onyx.static.util :refer [kw->fn exception?]]
            [onyx.types :refer [->Route]]))

(defn join-output-paths [all to-add downstream]
  (cond (= to-add :all) (set downstream)
        (= to-add :none) #{}
        :else (into (set all) to-add)))

(defn choose-output-paths
  [event compiled-flow-conditions result message downstream]
  (reduce
   (fn [{:keys [flow exclusions] :as all} entry]
     (cond ((:flow/predicate entry) [event (:message (:root result)) message (map :message (:leaves result))])
           (if (:flow/short-circuit? entry)
             (reduced (->Route (join-output-paths flow (:flow/to entry) downstream)
                               (into (set exclusions) (:flow/exclude-keys entry))
                               (:flow/post-transform entry)
                               (:flow/action entry)))
             (->Route (join-output-paths flow (:flow/to entry) downstream)
                      (into (set exclusions) (:flow/exclude-keys entry))
                      nil
                      nil))

           (= (:flow/action entry) :retry)
           (->Route (join-output-paths flow (:flow/to entry) downstream)
                    (into (set exclusions) (:flow/exclude-keys entry))
                    nil
                    nil)

           :else all))
   (->Route #{} #{} nil nil)
   compiled-flow-conditions))

(defn route-data
  [event {:keys [egress-ids compiled-ex-fcs compiled-norm-fcs flow-conditions] :as compiled} result message]
  (if (nil? flow-conditions)
    (if (exception? message)
      (let [e (:exception (ex-data message))]
        (lc/handle-exception
         event :lifecycle/apply-fn e (:compiled-handle-exception-fn compiled)))
      (->Route egress-ids nil nil nil))
    (if (exception? message)
      (if (seq compiled-ex-fcs)
        (choose-output-paths event compiled-ex-fcs result (:exception (ex-data message)) egress-ids)
        (throw (:exception (ex-data message))))
      (if (seq compiled-norm-fcs)
        (choose-output-paths event compiled-norm-fcs result message egress-ids)
        (->Route egress-ids nil nil nil)))))

(defn apply-post-transformation [message routes event]
  (let [post-transformation (:post-transformation routes)
        msg (if (and (exception? message) post-transformation)
              (let [data (ex-data message)
                    f (kw->fn post-transformation)]
                (f event (:segment data) (:exception data)))
              message)]
    (reduce dissoc msg (:exclusions routes))))

(defn flow-conditions-transform
  [message routes event compiled]
  (if (:flow-conditions compiled)
    (apply-post-transformation message routes event)
    message))<commit_msg>Fix flow conditions route IDs. Route no where when exception without handlers.<commit_after>(ns onyx.flow-conditions.fc-routing
  (:require [onyx.lifecycles.lifecycle-invoke :as lc]
            [onyx.static.util :refer [kw->fn exception?]]
            [onyx.types :refer [->Route]]))

(defn join-output-paths [all to-add downstream]
  (cond (= to-add :all) (set downstream)
        (= to-add :none) #{}
        :else (into (set all) to-add)))

(defn choose-output-paths
  [event compiled-flow-conditions result message downstream]
  (reduce
   (fn [{:keys [flow exclusions] :as all} entry]
     (cond ((:flow/predicate entry) [event (:message (:root result)) message (map :message (:leaves result))])
           (if (:flow/short-circuit? entry)
             (reduced (->Route (join-output-paths flow (:flow/to entry) downstream)
                               (into (set exclusions) (:flow/exclude-keys entry))
                               (:flow/post-transform entry)
                               (:flow/action entry)))
             (->Route (join-output-paths flow (:flow/to entry) downstream)
                      (into (set exclusions) (:flow/exclude-keys entry))
                      nil
                      nil))

           (= (:flow/action entry) :retry)
           (->Route (join-output-paths flow (:flow/to entry) downstream)
                    (into (set exclusions) (:flow/exclude-keys entry))
                    nil
                    nil)

           :else all))
   (->Route #{} #{} nil nil)
   compiled-flow-conditions))

(defn route-data
  [event {:keys [egress-ids compiled-ex-fcs compiled-norm-fcs flow-conditions] :as compiled} result message]
  (if (nil? flow-conditions)
    (if (exception? message)
      (let [e (:exception (ex-data message))]
        (lc/handle-exception
         event :lifecycle/apply-fn e (:compiled-handle-exception-fn compiled)))
      (->Route [] nil nil nil))
    (if (exception? message)
      (if (seq compiled-ex-fcs)
        (choose-output-paths event compiled-ex-fcs result (:exception (ex-data message)) egress-ids)
        (throw (:exception (ex-data message))))
      (if (seq compiled-norm-fcs)
        (choose-output-paths event compiled-norm-fcs result message egress-ids)
        (->Route egress-ids nil nil nil)))))

(defn apply-post-transformation [message routes event]
  (let [post-transformation (:post-transformation routes)
        msg (if (and (exception? message) post-transformation)
              (let [data (ex-data message)
                    f (kw->fn post-transformation)]
                (f event (:segment data) (:exception data)))
              message)]
    (reduce dissoc msg (:exclusions routes))))

(defn flow-conditions-transform
  [message routes event compiled]
  (if (:flow-conditions compiled)
    (apply-post-transformation message routes event)
    message))
{'fraction_non_alphanumeric': 0.13513513513513514, 'fraction_numerical': 0.0, 'mean_word_length': 2.259643062751871, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, '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': '3420344', 'n_tokens_mistral': 1684, 'n_tokens_neox': 1587, 'n_words': 459}