<commit_before>(ns onyx.windowing.triggers
  (:require [onyx.windowing.coerce :refer [to-standard-units]]
            [onyx.windowing.window-id :as wid]
            [onyx.peer.operation :refer [kw->fn]]
            [onyx.static.planning :refer [find-window]]
            [taoensso.timbre :refer [fatal]]))

(defmulti trigger-setup
  (fn [event trigger id]
    (:trigger/type trigger)))

(defmulti trigger-teardown
  (fn [event trigger id]
    (:trigger/type trigger)))

(defmulti refine-state
  (fn [event trigger]
    (:trigger/refinement trigger)))

(defmethod trigger-setup :periodically
  [event trigger id]
  ;; TODO: validate that :trigger/period is a time-based value.
  (let [f (kw->fn (:trigger/sync trigger))
        fut
        (future
          (loop []
            (try
              (let [ms (apply to-standard-units (:trigger/period trigger))]
                (Thread/sleep ms)
                (let [state (refine-state event trigger)
                      window-ids (get state (:trigger/window-id trigger))]
                  (doseq [[window-id state] window-ids]
                    (let [window (find-window (:onyx.core/windows event) (:trigger/window-id trigger))
                          win-min (or (:window/min-value window) 0)
                          w-range (apply to-standard-units (:window/range window))
                          w-slide (apply to-standard-units (:window/slide window))
                          lower (wid/extent-lower win-min w-range w-slide window-id)
                          upper (wid/extent-upper win-min w-slide window-id)]
                      (f event window-id lower upper state)))))
              (catch InterruptedException e
                (throw e))
              (catch Throwable e
                (fatal e)))
            (recur)))]
    (assoc-in event [:onyx.triggers/period-threads id] fut)))<commit_msg>Can't assume slide is there for fixed windows.<commit_after>(ns onyx.windowing.triggers
  (:require [onyx.windowing.coerce :refer [to-standard-units]]
            [onyx.windowing.window-id :as wid]
            [onyx.peer.operation :refer [kw->fn]]
            [onyx.static.planning :refer [find-window]]
            [taoensso.timbre :refer [fatal]]))

(defmulti trigger-setup
  (fn [event trigger id]
    (:trigger/type trigger)))

(defmulti trigger-teardown
  (fn [event trigger id]
    (:trigger/type trigger)))

(defmulti refine-state
  (fn [event trigger]
    (:trigger/refinement trigger)))

(defmethod trigger-setup :periodically
  [event trigger id]
  ;; TODO: validate that :trigger/period is a time-based value.
  (let [f (kw->fn (:trigger/sync trigger))
        fut
        (future
          (loop []
            (try
              (let [ms (apply to-standard-units (:trigger/period trigger))]
                (Thread/sleep ms)
                (let [state (refine-state event trigger)
                      window-ids (get state (:trigger/window-id trigger))]
                  (doseq [[window-id state] window-ids]
                    (let [window (find-window (:onyx.core/windows event) (:trigger/window-id trigger))
                          win-min (or (:window/min-value window) 0)
                          w-range (apply to-standard-units (:window/range window))
                          w-slide (apply to-standard-units (or (:window/slide window) (:window/range window)))
                          lower (wid/extent-lower win-min w-range w-slide window-id)
                          upper (wid/extent-upper win-min w-slide window-id)]
                      (f event window-id lower upper state)))))
              (catch InterruptedException e
                (throw e))
              (catch Throwable e
                (fatal e)))
            (recur)))]
    (assoc-in event [:onyx.triggers/period-threads id] fut)))
{'fraction_non_alphanumeric': 0.12398213816653533, 'fraction_numerical': 0.0005253480430785396, 'mean_word_length': 1.8481675392670156, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '5808646', 'n_tokens_mistral': 1117, 'n_tokens_neox': 1068, 'n_words': 291}