<commit_before>(comment 
   Celestial, Copyright 2012 <NAME>, <EMAIL>
   Licensed under the Apache License,
   Version 2.0  (the "License") you may not use this file except in compliance with the License.
   You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.)

(ns celestial.jobs
  (:refer-clojure :exclude [identity])
  (:use   
    [gelfino.timbre :only (set-tid)]
    [celestial.common :only (get*)]
    [flatland.useful.map :on map-vals]
    [clojure.core.strint :only (<<)]
    [celestial.common :only (minute import-logging)]
    [celestial.redis :only (create-worker wcar with-lock)]
    [taoensso.timbre :only (debug info error warn trace)]
    [celestial.workflows :only (reload destroy puppetize full-cycle run-action)]) 
  (:require  
    [taoensso.carmine :as car]
    [taoensso.carmine.message-queue :as mq]))

(import-logging)

(def workers (atom {}))

(def defaults {:wait-time 5 :expiry 30})

(defn job-exec [f {:keys [identity args tid] :as spec}]
  "Executes a job function tries to lock identity first (if used)"
  (debug tid)
  (set-tid tid 
    (try 
      (if identity
        (with-lock identity 
         (fn [] (apply f args) :success)
         (map-vals (or (get* :job) defaults) #(* minute %) )) 
        (do (apply f args) :success))
      (catch Throwable e (error e) :error))))

(def jobs 
  (atom 
    {:reload [reload 2] :destroy [destroy 2] :provision [puppetize 2]
     :stage [full-cycle 2] :run-action [run-action 2] }))

(defn create-wks [queue f total]
  "create a count of workers for queue"
  (mapv (fn [v] (create-worker (name queue) (partial job-exec f))) (range total)))

(defn initialize-workers []
  (dosync 
    (doseq [[q [f c]] @jobs]
      (swap! workers assoc q (create-wks q f c)))))

(defn clear-all []
  (doseq [q (map name (keys @jobs))]
    (trace (<< "clearing ~{q} queue"))<commit_msg>clearing debug<commit_after>(comment 
   Celestial, Copyright 2012 <NAME>, <EMAIL>
   Licensed under the Apache License,
   Version 2.0  (the "License") you may not use this file except in compliance with the License.
   You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.)

(ns celestial.jobs
  (:refer-clojure :exclude [identity])
  (:use   
    [gelfino.timbre :only (set-tid)]
    [celestial.common :only (get*)]
    [flatland.useful.map :on map-vals]
    [clojure.core.strint :only (<<)]
    [celestial.common :only (minute import-logging)]
    [celestial.redis :only (create-worker wcar with-lock)]
    [taoensso.timbre :only (debug info error warn trace)]
    [celestial.workflows :only (reload destroy puppetize full-cycle run-action)]) 
  (:require  
    [taoensso.carmine :as car]
    [taoensso.carmine.message-queue :as mq]))

(import-logging)

(def workers (atom {}))

(def defaults {:wait-time 5 :expiry 30})

(defn job-exec [f {:keys [identity args tid] :as spec}]
  "Executes a job function tries to lock identity first (if used)"
  (set-tid tid 
    (try 
      (if identity
        (with-lock identity 
         (fn [] (apply f args) :success)
         (map-vals (or (get* :job) defaults) #(* minute %) )) 
        (do (apply f args) :success))
      (catch Throwable e (error e) :error))))

(def jobs 
  (atom 
    {:reload [reload 2] :destroy [destroy 2] :provision [puppetize 2]
     :stage [full-cycle 2] :run-action [run-action 2] }))

(defn create-wks [queue f total]
  "create a count of workers for queue"
  (mapv (fn [v] (create-worker (name queue) (partial job-exec f))) (range total)))

(defn initialize-workers []
  (dosync 
    (doseq [[q [f c]] @jobs]
      (swap! workers assoc q (create-wks q f c)))))

(defn clear-all []
  (doseq [q (map name (keys @jobs))]
    (trace (<< "clearing ~{q} queue"))
{'fraction_non_alphanumeric': 0.12931231437057344, 'fraction_numerical': 0.00731094356865433, 'mean_word_length': 3.3346534653465345, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '733970', 'n_tokens_mistral': 1495, 'n_tokens_neox': 1418, 'n_words': 520}