<commit_before>(ns taoensso.carmine
  "Clojure Redis client & message queue."
  {:author "<NAME>"}
  (:refer-clojure :exclude [time get set keys type sync sort eval])
  (:require [clojure.string :as str]
            [taoensso.carmine
             (utils       :as utils)
             (protocol    :as protocol)
             (connections :as conns)
             (commands    :as commands)]
            [taoensso.timbre :as timbre])
  (:import [org.apache.commons.pool.impl GenericKeyedObjectPool]
           [taoensso.carmine.connections ConnectionPool]
           [taoensso.carmine.protocol    Frozen Raw]
           java.net.URI))

;;;; Connections

(defn make-conn-pool
  "For option documentation see http://goo.gl/EiTbn"
  [& options]
  (let [;; Defaults adapted from Jedis
        defaults {:test-while-idle?              true
                  :num-tests-per-eviction-run    -1
                  :min-evictable-idle-time-ms    60000
                  :time-between-eviction-runs-ms 30000}]
    (ConnectionPool.
     (reduce conns/set-pool-option
             (GenericKeyedObjectPool. (conns/make-connection-factory))
             (merge defaults (apply hash-map options))))))

(defn- parse-uri [uri]
  (when uri
    (let [^URI uri (if (instance? URI uri) uri (URI. uri))
          [user password] (.split (str (.getUserInfo uri)) ":")
          port (.getPort uri)]
      (-> {:host (.getHost uri)}
          (#(if (pos? port) (assoc % :port     port)     %))
          (#(if password    (assoc % :password password) %))))))

(comment (parse-uri "redis://redistogo:pass@panga.redistogo.com:9475/"))

(defn make-conn-spec
  [& {:keys [uri host port password timeout-ms db] :as opts}]
  (let [defaults {:host "127.0.0.1" :port 6379}
        opts     (if-let [timeout (:timeout opts)] ; Support deprecated opt
                   (assoc opts :timeout-ms timeout)
                   opts)]
    (merge defaults opts (parse-uri uri))))

(defmacro with-conn
  "Evaluates body in the context of a thread-bound pooled connection to a Redis
  server. Sends Redis commands to server as pipeline and returns the server's
  response. Releases connection back to pool when done.

  Use `make-conn-pool` and `make-conn-spec` to generate the required arguments."
  [connection-pool connection-spec & body]
  `(let [pool# (or ~connection-pool conns/non-pooled-connection-pool)
         spec# (or ~connection-spec (make-conn-spec))
         conn# (try (conns/get-conn pool# spec#)
                    (catch Exception e#
                      (throw (Exception. "Carmine connection error" e#))))]
     (try
       (let [response# (protocol/with-context conn# ~@body)]
         (conns/release-conn pool# conn#) response#)
       (catch Exception e# (conns/release-conn pool# conn# e#) (throw e#)))))

;;;; Misc

;;; Note (number? x) checks for backwards compatibility with pre-v0.11 Carmine
;;; versions that auto-serialized simple number types
(defn as-long   [x] (when x (if (number? x) (long   x) (Long/parseLong     x))))
(defn as-double [x] (when x (if (number? x) (double x) (Double/parseDouble x))))
(defn as-bool   [x] (when x
                      (cond (or (true? x) (false? x))            x
                            (or (= x "false") (= x "0") (= x 0)) false
                            (or (= x "true")  (= x "1") (= x 1)) true
                            :else
                            (throw (Exception. (str "Couldn't coerce as bool: "
                                                    x))))))

(defmacro with-parser
  "Alpha - subject to change!!
  Wraps body so that replies to any wrapped Redis commands will be parsed with
  `(f reply)`. Replaces any current parser; removes parser when `f` is nil."
  [f & body] `(binding [protocol/*parser* ~f] ~@body))

(defmacro parse-long    [& body] `(with-parser as-long   ~@body))
(defmacro parse-double  [& body] `(with-parser as-double ~@body))
(defmacro parse-bool    [& body] `(with-parser as-bool   ~@body))
(defmacro parse-keyword [& body] `(with-parser keyword   ~@body))

(defn kname
  "Joins keywords, integers, and strings to form an idiomatic compound Redis key
  name.

  Suggested key naming style:
    * \"category:subcategory:id:field\" basic form.
    * Singular category names (\"account\" rather than \"accounts\").
    * Dashes for long names (\"email-address\" rather than \"emailAddress\", etc.)."

  [& parts] (str/join ":" (map utils/keyname (filter identity parts))))

(comment (kname :foo/bar :baz "qux" nil 10))

(defn freeze
  "Forces argument of any type (including simple number and binary types) to be
  subject to automatic de/serialization."
  [x] (protocol/Frozen. x))

(defn raw "Alpha - subject to change." [x] (protocol/Raw. x))
(defmacro parse-raw "Alpha - subject to change."
  [& body] `(with-parser (with-meta identity {:raw? true}) ~@body))

(defn return
  "Alpha - subject to change.
  Special command that takes any value and returns it unchanged as part of
  an enclosing `with-conn` pipeline response."
  [value]
  (let [vfn (constantly value)
        parser protocol/*parser*]
    (swap! (:parser-queue protocol/*context*) conj
           (with-meta (if parser (comp parser vfn) vfn)
             {:dummy-reply? true}))))

(comment (wc (return :foo) (ping) (return :bar))
         (wc (with-parser name (return :foo)) (ping) (return :bar)))

(defmacro with-replies
  "Alpha - subject to change!!
  Evaluates body, immediately returning the server's response to any contained
  Redis commands (i.e. before enclosing `with-conn` ends). Ignores any parser
  in enclosing (not _enclosed_) context.

  As an implementation detail, stashes and then `return`s any replies already
  queued with Redis server: i.e. should be compatible with pipelining."
  {:arglists '([:as-pipeline & body] [& body])}
  [& [s1 & sn :as sigs]]
  (let [as-pipeline? (= s1 :as-pipeline)
        body (if as-pipeline? sn sigs)]
    `(let [stashed-replies# (protocol/get-replies! true)]
       (try (with-parser nil ~@body) ; Herewith dragons; tread lightly
            (protocol/get-replies! ~as-pipeline?)
            (finally
             ;; doseq here broken with Clojure <1.5, Ref. http://goo.gl/5DvRt
             (with-parser nil (dorun (map return stashed-replies#))))))))

(comment (wc (echo 1) (println (with-replies (ping))) (echo 2))
         (wc (echo 1) (println (with-replies :as-pipeline (ping))) (echo 2)))

;;;; Standard commands

(commands/defcommands) ; This kicks ass - big thanks to <NAME>!

;;;; Helper commands

(def hash-script
  (memoize
   (fn [script]
     (org.apache.commons.codec.digest.DigestUtils/shaHex (str script)))))

(defn evalsha*
  "Like `evalsha` but automatically computes SHA1 hash for script."
  [script numkeys key & args]
  (apply evalsha (hash-script script) numkeys key args))

(defn eval*
  "Optimistically tries to send `evalsha` command for given script. In the event
  of a \"NOSCRIPT\" reply, reattempts with `eval`. Returns the final command's
  reply."
  [script numkeys key & args]
  (let [parser protocol/*parser*
        [r & _] (->> (apply evalsha* script numkeys key args)
                     (with-replies :as-pipeline))]
    (if (instance? Exception r)
      (if (.startsWith (.getMessage ^Exception r) "NOSCRIPT")
        (with-parser parser (apply eval script numkeys key args))
        (throw r))
      (with-parser parser (return r)))))

(def ^:private interpolate-script<commit_msg>Allow pipelined (non-throwing) Lua script exceptions<commit_after>(ns taoensso.carmine
  "Clojure Redis client & message queue."
  {:author "<NAME>"}
  (:refer-clojure :exclude [time get set keys type sync sort eval])
  (:require [clojure.string :as str]
            [taoensso.carmine
             (utils       :as utils)
             (protocol    :as protocol)
             (connections :as conns)
             (commands    :as commands)]
            [taoensso.timbre :as timbre])
  (:import [org.apache.commons.pool.impl GenericKeyedObjectPool]
           [taoensso.carmine.connections ConnectionPool]
           [taoensso.carmine.protocol    Frozen Raw]
           java.net.URI))

;;;; Connections

(defn make-conn-pool
  "For option documentation see http://goo.gl/EiTbn"
  [& options]
  (let [;; Defaults adapted from Jedis
        defaults {:test-while-idle?              true
                  :num-tests-per-eviction-run    -1
                  :min-evictable-idle-time-ms    60000
                  :time-between-eviction-runs-ms 30000}]
    (ConnectionPool.
     (reduce conns/set-pool-option
             (GenericKeyedObjectPool. (conns/make-connection-factory))
             (merge defaults (apply hash-map options))))))

(defn- parse-uri [uri]
  (when uri
    (let [^URI uri (if (instance? URI uri) uri (URI. uri))
          [user password] (.split (str (.getUserInfo uri)) ":")
          port (.getPort uri)]
      (-> {:host (.getHost uri)}
          (#(if (pos? port) (assoc % :port     port)     %))
          (#(if password    (assoc % :password password) %))))))

(comment (parse-uri "redis://redistogo:pass@panga.redistogo.com:9475/"))

(defn make-conn-spec
  [& {:keys [uri host port password timeout-ms db] :as opts}]
  (let [defaults {:host "127.0.0.1" :port 6379}
        opts     (if-let [timeout (:timeout opts)] ; Support deprecated opt
                   (assoc opts :timeout-ms timeout)
                   opts)]
    (merge defaults opts (parse-uri uri))))

(defmacro with-conn
  "Evaluates body in the context of a thread-bound pooled connection to a Redis
  server. Sends Redis commands to server as pipeline and returns the server's
  response. Releases connection back to pool when done.

  Use `make-conn-pool` and `make-conn-spec` to generate the required arguments."
  [connection-pool connection-spec & body]
  `(let [pool# (or ~connection-pool conns/non-pooled-connection-pool)
         spec# (or ~connection-spec (make-conn-spec))
         conn# (try (conns/get-conn pool# spec#)
                    (catch Exception e#
                      (throw (Exception. "Carmine connection error" e#))))]
     (try
       (let [response# (protocol/with-context conn# ~@body)]
         (conns/release-conn pool# conn#) response#)
       (catch Exception e# (conns/release-conn pool# conn# e#) (throw e#)))))

;;;; Misc

;;; Note (number? x) checks for backwards compatibility with pre-v0.11 Carmine
;;; versions that auto-serialized simple number types
(defn as-long   [x] (when x (if (number? x) (long   x) (Long/parseLong     x))))
(defn as-double [x] (when x (if (number? x) (double x) (Double/parseDouble x))))
(defn as-bool   [x] (when x
                      (cond (or (true? x) (false? x))            x
                            (or (= x "false") (= x "0") (= x 0)) false
                            (or (= x "true")  (= x "1") (= x 1)) true
                            :else
                            (throw (Exception. (str "Couldn't coerce as bool: "
                                                    x))))))

(defmacro with-parser
  "Alpha - subject to change!!
  Wraps body so that replies to any wrapped Redis commands will be parsed with
  `(f reply)`. Replaces any current parser; removes parser when `f` is nil."
  [f & body] `(binding [protocol/*parser* ~f] ~@body))

(defmacro parse-long    [& body] `(with-parser as-long   ~@body))
(defmacro parse-double  [& body] `(with-parser as-double ~@body))
(defmacro parse-bool    [& body] `(with-parser as-bool   ~@body))
(defmacro parse-keyword [& body] `(with-parser keyword   ~@body))

(defn kname
  "Joins keywords, integers, and strings to form an idiomatic compound Redis key
  name.

  Suggested key naming style:
    * \"category:subcategory:id:field\" basic form.
    * Singular category names (\"account\" rather than \"accounts\").
    * Dashes for long names (\"email-address\" rather than \"emailAddress\", etc.)."

  [& parts] (str/join ":" (map utils/keyname (filter identity parts))))

(comment (kname :foo/bar :baz "qux" nil 10))

(defn freeze
  "Forces argument of any type (including simple number and binary types) to be
  subject to automatic de/serialization."
  [x] (protocol/Frozen. x))

(defn raw "Alpha - subject to change." [x] (protocol/Raw. x))
(defmacro parse-raw "Alpha - subject to change."
  [& body] `(with-parser (with-meta identity {:raw? true}) ~@body))

(defn return
  "Alpha - subject to change.
  Special command that takes any value and returns it unchanged as part of
  an enclosing `with-conn` pipeline response."
  [value]
  (let [vfn (constantly value)
        parser protocol/*parser*]
    (swap! (:parser-queue protocol/*context*) conj
           (with-meta (if parser (comp parser vfn) vfn)
             {:dummy-reply? true}))))

(comment (wc (return :foo) (ping) (return :bar))
         (wc (with-parser name (return :foo)) (ping) (return :bar)))

(defmacro with-replies
  "Alpha - subject to change!!
  Evaluates body, immediately returning the server's response to any contained
  Redis commands (i.e. before enclosing `with-conn` ends). Ignores any parser
  in enclosing (not _enclosed_) context.

  As an implementation detail, stashes and then `return`s any replies already
  queued with Redis server: i.e. should be compatible with pipelining."
  {:arglists '([:as-pipeline & body] [& body])}
  [& [s1 & sn :as sigs]]
  (let [as-pipeline? (= s1 :as-pipeline)
        body (if as-pipeline? sn sigs)]
    `(let [stashed-replies# (protocol/get-replies! true)]
       (try (with-parser nil ~@body) ; Herewith dragons; tread lightly
            (protocol/get-replies! ~as-pipeline?)
            (finally
             ;; doseq here broken with Clojure <1.5, Ref. http://goo.gl/5DvRt
             (with-parser nil (dorun (map return stashed-replies#))))))))

(comment (wc (echo 1) (println (with-replies (ping))) (echo 2))
         (wc (echo 1) (println (with-replies :as-pipeline (ping))) (echo 2)))

;;;; Standard commands

(commands/defcommands) ; This kicks ass - big thanks to <NAME>!

;;;; Helper commands

(def hash-script
  (memoize
   (fn [script]
     (org.apache.commons.codec.digest.DigestUtils/shaHex (str script)))))

(defn evalsha*
  "Like `evalsha` but automatically computes SHA1 hash for script."
  [script numkeys key & args]
  (apply evalsha (hash-script script) numkeys key args))

(defn eval*
  "Optimistically tries to send `evalsha` command for given script. In the event
  of a \"NOSCRIPT\" reply, reattempts with `eval`. Returns the final command's
  reply."
  [script numkeys key & args]
  (let [parser protocol/*parser*
        [r & _] (->> (apply evalsha* script numkeys key args)
                     (with-replies :as-pipeline))]
    (with-parser parser
      (if (and (instance? Exception r)
               (.startsWith (.getMessage ^Exception r) "NOSCRIPT"))
        (apply eval script numkeys key args)
        (return r)))))

(def ^:private interpolate-script
{'fraction_non_alphanumeric': 0.1364921254543007, 'fraction_numerical': 0.005922735226813838, 'mean_word_length': 2.6232626188734454, 'pattern_counts': {'":': 4, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5291561', 'n_tokens_mistral': 4805, 'n_tokens_neox': 4632, 'n_words': 1536}