<commit_before>(ns onyx.peer.queryable-state-manager
  (:require [clojure.core.async :refer [chan close! poll! >!!]]
            [onyx.state.serializers.utils]
            [com.stuartsierra.component :as component]
            [onyx.peer.window-state :as ws]
            [onyx.state.protocol.db :as db]
            [taoensso.timbre :refer [fatal info]])
  (:import [java.util.concurrent.locks LockSupport]))

(defn state-key [replica-version event]
  [(:onyx.core/job-id event)
   replica-version
   (:onyx.core/task event)
   (:onyx.core/slot-id event)])

(defn add-store [state peer-config [_ replica-version event exported]]
  (let [serializers (onyx.state.serializers.utils/event->state-serializers event)]
    (swap! state 
           assoc 
           (state-key replica-version event)
           {:state-indexes (ws/state-indexes event)
            :db (db/open-db-reader peer-config exported serializers)})))

(defn drop-store [state peer-config [_ replica-version event]]
  (let [serializers (onyx.state.serializers.utils/event->state-serializers event)
        k (state-key replica-version event)
        store (get @state k)]
    (db/close! store)
    (swap! state dissoc k))) 

(defn processing-loop [peer-config shutdown state ch]
  (loop []
    (when-not @shutdown
      (if-let [cmd (poll! ch)]
        (case (first cmd)
          :created-db (add-store state peer-config cmd)
          :drop-db (drop-store state peer-config cmd))
         (LockSupport/parkNanos (* 10 1000000)))
      (recur))))

(defrecord OnyxStateStoreGroup [peer-config ch state shutdown]
  component/Lifecycle<commit_msg>Disable storing queryable state until production ready.<commit_after>(ns onyx.peer.queryable-state-manager
  (:require [clojure.core.async :refer [chan close! poll! >!!]]
            [onyx.state.serializers.utils]
            [com.stuartsierra.component :as component]
            [onyx.peer.window-state :as ws]
            [onyx.state.protocol.db :as db]
            [taoensso.timbre :refer [fatal info]])
  (:import [java.util.concurrent.locks LockSupport]))

(defn state-key [replica-version event]
  [(:onyx.core/job-id event)
   replica-version
   (:onyx.core/task event)
   (:onyx.core/slot-id event)])

(defn add-store [state peer-config [_ replica-version event exported]]
  #_(let [serializers (onyx.state.serializers.utils/event->state-serializers event)]
    (swap! state 
           assoc 
           (state-key replica-version event)
           {:state-indexes (ws/state-indexes event)
            :db (db/open-db-reader peer-config exported serializers)})))

(defn drop-store [state peer-config [_ replica-version event]]
  #_(let [serializers (onyx.state.serializers.utils/event->state-serializers event)
        k (state-key replica-version event)
        store (get @state k)]
    (db/close! store)
    (swap! state dissoc k))) 

(defn processing-loop [peer-config shutdown state ch]
  (loop []
    (when-not @shutdown
      (if-let [cmd (poll! ch)]
        (case (first cmd)
          :created-db (add-store state peer-config cmd)
          :drop-db (drop-store state peer-config cmd))
         (LockSupport/parkNanos (* 10 1000000)))
      (recur))))

(defrecord OnyxStateStoreGroup [peer-config ch state shutdown]
  component/Lifecycle
{'fraction_non_alphanumeric': 0.13380281690140844, 'fraction_numerical': 0.005511328842620943, 'mean_word_length': 2.8708530805687205, 'pattern_counts': {'":': 0, '<': 3, '<?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': '6634594', 'n_tokens_mistral': 1087, 'n_tokens_neox': 1033, 'n_words': 249}