<commit_before>(ns group-redis.core
  (:require [fun-utils.core :refer [fixdelay stop-fixdelay]]
            [clojure.core.async :refer [chan >! <! >!! go mult tap untap sliding-buffer close!]]
            [taoensso.carmine :as car :refer [wcar]])
  (:import [java.net InetAddress]))


;Paths
; /[group-name]/members/[node-name]
; /[group-name]/empherals/[path..]/...
; /[group-name]/locks/[path..]/...

(defonce host-name (-> (InetAddress/getLocalHost) (.getHostName)))

(defn members-path 
  ([connector member]
    (clojure.string/join "/" [(members-path connector) member]))
  ([{:keys [group-name]}]
  (clojure.string/join ["/" (clojure.string/join "/" [group-name "members"])])))


(defn persistent-path [{:keys [group-name]} & path]
  (clojure.string/join ["/" (clojure.string/join "/" (flatten [group-name "persistent" path]))]))

(defn empherals-path [{:keys [group-name]} & path]
  (clojure.string/join ["/" (clojure.string/join "/" (flatten [group-name "empherals" path]))]))

(defn lock-path [{:keys [group-name]} & path]
  (clojure.string/join ["/" (clojure.string/join "/" (flatten [group-name "locks" path]))]))


(defn close [{:keys [state-ref member-event-ch heart-beat-ch]}]
  ;delete connection's state
  (stop-fixdelay heart-beat-ch)
  (close! member-event-ch)
  
  (dosync (alter state-ref (fn [state] {}))))
                        
(defn create-connection 
  "Creates a redis connection the default port used is 6379"
  ([host]
    (create-connection host {}))
  ([host {:keys [port max-active password timeout] :or {port 6379 max-active 20 timeout 4000}}]
                         {:pool {:max-active max-active}
                          :spec {:host  host
                               :port    port
                               :password password
                               :timeout  timeout}}))


(defn calc-ttl [heart-beat-freq]
  "Takes the heart-beat-freq and calculates the time to live relative to the heart beat"
      (long (+ heart-beat-freq (/ heart-beat-freq 2))))


(defn persistent-get [{:keys [conn state-ref] :as connector} path]
  (let [final-path (persistent-path connector path)]
    (car/wcar conn (car/get final-path))))

(defn persistent-set [{:keys [conn state-ref] :as connector} path val]
  (let [final-path (persistent-path connector path)]
    (car/wcar conn (car/set final-path val))))


(defn persistent-set* [{:keys [conn state-ref] :as connector} data]
  "Set multipl values, data should be [[path val]... ]"
    (car/wcar conn
              (doseq [[path val] data]
                (car/set (persistent-path connector path) val))))
  
(defn empheral-set [{:keys [conn state-ref conf] :as connector} path val]
  (let [final-path (empherals-path connector path)
        {:keys [heart-beat-freq]} conf
        expire (calc-ttl heart-beat-freq)]
    (dosync (alter state-ref (fn [state] (assoc state :empherals (into #{} (conj (:empherals state) {:path final-path :val val}))))))
    (car/wcar conn 
              (do
                (car/set final-path val)
                (car/expire final-path expire)
              ))))

(defn empheral-get [connector path ]
  (->> connector :state-ref deref :empherals (filter #(= (:path %) (empherals-path connector path) )) first :val))

(defn release 
  "Releases the lock, return true if the lock was released, false otherwise"
  ([connecor path]
    (release connecor host-name path))
  ([{:keys [conn conf state-ref] :as connector} member path]
  (let [lock-path (lock-path connector path)
        lock-val (car/wcar conn (car/get lock-path))]
    
    (if (= (:member lock-val) member)
      (do 
        ;remove from state
        (try (dosync (alter state-ref (fn [state]
                                   (assoc state :locks (filter (complement #(= (:path %) lock-path)) (:locks state))))))
          (finally ;delete from redis
                 (car/wcar conn (car/del lock-path))))
        true)
      false))))
      
  
(defn lock <commit_msg>update logging<commit_after>(ns group-redis.core
  (:require [fun-utils.core :refer [fixdelay stop-fixdelay]]
            [clojure.core.async :refer [chan >! <! >!! go mult tap untap sliding-buffer close!]]
            [clojure.tools.logging :refer [info]]
            [taoensso.carmine :as car :refer [wcar]])
  (:import [java.net InetAddress]))


;Paths
; /[group-name]/members/[node-name]
; /[group-name]/empherals/[path..]/...
; /[group-name]/locks/[path..]/...

(defonce host-name (-> (InetAddress/getLocalHost) (.getHostName)))

(defn members-path 
  ([connector member]
    (clojure.string/join "/" [(members-path connector) member]))
  ([{:keys [group-name]}]
  (clojure.string/join ["/" (clojure.string/join "/" [group-name "members"])])))


(defn persistent-path [{:keys [group-name]} & path]
  (clojure.string/join ["/" (clojure.string/join "/" (flatten [group-name "persistent" path]))]))

(defn empherals-path [{:keys [group-name]} & path]
  (clojure.string/join ["/" (clojure.string/join "/" (flatten [group-name "empherals" path]))]))

(defn lock-path [{:keys [group-name]} & path]
  (clojure.string/join ["/" (clojure.string/join "/" (flatten [group-name "locks" path]))]))


(defn close [{:keys [state-ref member-event-ch heart-beat-ch]}]
  ;delete connection's state
  (stop-fixdelay heart-beat-ch)
  (close! member-event-ch)
  
  (dosync (alter state-ref (fn [state] {}))))
                        
(defn create-connection 
  "Creates a redis connection the default port used is 6379"
  ([host]
    (create-connection host {}))
  ([host {:keys [port max-active password timeout] :or {port 6379 max-active 20 timeout 4000}}]
                         {:pool {:max-active max-active}
                          :spec {:host  host
                               :port    port
                               :password password
                               :timeout  timeout}}))


(defn calc-ttl [heart-beat-freq]
  "Takes the heart-beat-freq and calculates the time to live relative to the heart beat"
      (long (+ heart-beat-freq (/ heart-beat-freq 2))))


(defn persistent-get [{:keys [conn state-ref] :as connector} path]
  (let [final-path (persistent-path connector path)]
    (car/wcar conn (car/get final-path))))

(defn persistent-set [{:keys [conn state-ref] :as connector} path val]
  (let [final-path (persistent-path connector path)]
    (car/wcar conn (car/set final-path val))))


(defn persistent-set* [{:keys [conn state-ref] :as connector} data]
  "Set multipl values, data should be [[path val]... ]"
    (car/wcar conn
              (doseq [[path val] data]
                (car/set (persistent-path connector path) val))))
  
(defn empheral-set [{:keys [conn state-ref conf] :as connector} path val]
  (let [final-path (empherals-path connector path)
        {:keys [heart-beat-freq]} conf
        expire (calc-ttl heart-beat-freq)]
    (dosync (alter state-ref (fn [state] (assoc state :empherals (into #{} (conj (:empherals state) {:path final-path :val val}))))))
    (car/wcar conn 
              (do
                (car/set final-path val)
                (car/expire final-path expire)
              ))))

(defn empheral-get [connector path ]
  (->> connector :state-ref deref :empherals (filter #(= (:path %) (empherals-path connector path) )) first :val))

(defn release 
  "Releases the lock, return true if the lock was released, false otherwise"
  ([connecor path]
    (release connecor host-name path))
  ([{:keys [conn conf state-ref] :as connector} member path]
  (let [lock-path (lock-path connector path)
        lock-val (car/wcar conn (car/get lock-path))]
    
    ;(info "Release path " path " lock-path " lock-path " lock-val " lock-val)
    (if (= (:member lock-val) member)
      (do 
        ;remove from state
        (try (dosync (alter state-ref (fn [state]
                                   (assoc state :locks (filter (complement #(= (:path %) lock-path)) (:locks state))))))
          (finally ;delete from redis
                 (car/wcar conn (car/del lock-path))))
        true)
      false))))
      
  
(defn lock 
{'fraction_non_alphanumeric': 0.1623166790952026, 'fraction_numerical': 0.0037285607755406414, 'mean_word_length': 2.5988372093023258, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3677932', 'n_tokens_mistral': 2741, 'n_tokens_neox': 2581, 'n_words': 722}