Dynamically generating queue functions from Jedis' source code

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "vendor/jedis"]
+	path = vendor/jedis
+	url = https://github.com/moclojer/jedis.git

Makefile

@@ -0,0 +1,5 @@
+jedis:
+	git submodule update --init --recursive --remote
+	cd vendor/jedis && make mvn-package-no-tests
+
+all: jedis

README.md

@@ -23,6 +23,12 @@ com.moclojer/rq {:mvn/version "0.x.x"}
 
 > see the versions distributed on clojars
 
+## building from source
+
+We build jedis ourselves, in order to build the queue functions directly from reflection.
+
+Simply run `make jedis`, and the library is be built and ready to be linked.
+
 ## example
 
 ```clojure

deps.edn

@@ -1,8 +1,11 @@
-{:paths   ["src"]
+{:paths ["src" "resources"]
  :deps
- {redis.clients/jedis {:mvn/version "5.1.2"}
+ {redis.clients/jedis {#_#_:mvn/version "5.1.2"
+                       :local/root "vendor/jedis/target/jedis-5.2.0-SNAPSHOT.jar"}
   org.clojure/tools.logging {:mvn/version "1.3.0"}
-  ch.qos.logback/logback-classic {:mvn/version "1.5.6"}}
+  ch.qos.logback/logback-classic {:mvn/version "1.5.6"}
+  camel-snake-kebab/camel-snake-kebab {:mvn/version "0.4.3"}
+  org.clojure/data.json {:mvn/version "2.5.0"}}
 
  :aliases
  {;; clj -A:dev -m com.moclojer.rq

resources/command-allowlist.edn

@@ -0,0 +1,4 @@
+#{"lpush" "rpush" "lpop" "rpop"  "brpop"
+  "blpop" "lrange" "lindex" "lset" "lrem"
+  "llen" "linsert" "ltrim" "rpoplpush"
+  "brpoplpush"  "lmove"}

src/com/moclojer/internal/reflection.clj

@@ -0,0 +1,81 @@
+(ns com.moclojer.internal.reflection
+  (:require
+   [camel-snake-kebab.core :as csk]
+   [clojure.string :as str]
+   [com.moclojer.rq.utils :as utils]))
+
+(defn unpack-parameter
+  [parameter]
+  {:type (.. parameter getType getName)
+   :name (csk/->kebab-case (.getName parameter))})
+
+(defn unpack-method
+  [method]
+  {:name (csk/->kebab-case (.getName method))
+   :parameters (map unpack-parameter (.getParameters method))})
+
+(defn reduce-method-overloads
+  [methods]
+  (reduce
+   (fn [overloaded-methods {:keys [name parameters]}]
+     (let [overload-count (count
+                           (filter
+                            #(str/starts-with? (key %) name)
+                            overloaded-methods))
+           cur-overload-id (when (> overload-count 0) overload-count)]
+       (assoc overloaded-methods (str name cur-overload-id) parameters)))
+   {} methods))
+
+(defn get-klazz-methods
+  [klazz allowlist]
+  (->> (.getMethods klazz)
+       (map unpack-method)
+       (filter #(contains? allowlist (:name %)))
+       (reduce-method-overloads)))
+
+(defmacro ->wrap-method
+  [method parameters]
+  (let [wrapped-method (clojure.string/replace method #"[`0-9]" "")
+        param-syms (map #(-> % :name symbol) parameters)]
+    `(defn ~(symbol method)
+       ~(str "Wraps redis.clients.jedis.JedisPooled." wrapped-method)
+
+       ~(-> (into ['client] param-syms)
+            (conj '& 'options))
+
+       (let [~{:keys ['pattern 'encoding 'decoding]
+               :or {'pattern :rq
+                    'encoding :edn
+                    'decoding :edn}} ~'options
+             ~'result ~(seq
+                        (into
+                         [(symbol (str "." wrapped-method)) '@client]
+                         (reduce
+                          (fn [acc par]
+                            (->> (cond
+                                   (= par 'key)
+                                   `(com.moclojer.rq.utils/pack-pattern
+                                     ~'pattern ~par)
+
+                                   (= par 'value)
+                                   `(com.moclojer.rq.utils/encode
+                                     ~'encoding ~par)
+
+                                   :else par)
+                                 (conj acc)))
+                          []
+                          param-syms)))]
+         (try
+           (com.moclojer.rq.utils/decode ~'decoding ~'result)
+           (catch ~'Exception ~'e
+             (.printStackTrace ~'e)
+             ~'result))))))
+
+(comment
+  (let [[method parameters] (-> redis.clients.jedis.JedisPooled
+                                (get-klazz-methods #{"lpush"})
+                                first)]
+    (clojure.pprint/pprint
+     (macroexpand-1 `(->wrap-method ~method ~parameters))))
+  ;;
+  )

src/com/moclojer/rq/pubsub.clj

@@ -28,7 +28,7 @@
       (do
         (log/warn "published message, but didn't meet min consumers. archiving..."
                   debug-args)
-        (queue/push! client channel message :pattern :pending)))
+        (queue/lpush client channel message :pattern :pending)))
 
     consumer-met?))
 
@@ -64,9 +64,7 @@
   on each of them."
   [client channel on-msg-fn]
   (loop [message-count 0]
-    (if-let [?message (queue/pop! client channel
-                                  :direction :r
-                                  :pattern :pending)]
+    (if-let [?message (queue/rpop client channel 1 :pattern :pending)]
       (do
         (try
           (on-msg-fn ?message)

src/com/moclojer/rq/queue.clj

@@ -1,73 +1,16 @@
 (ns com.moclojer.rq.queue
   (:refer-clojure :exclude [pop!])
   (:require
-   [clojure.edn :as edn]
-   [clojure.tools.logging :as log]
-   [com.moclojer.rq.utils :as utils]))
-
-(defn push!
-  "Push a message into a queue.
-
-  Options:
-
-  - direction: Direction to push the message (:l or :r)
-  - pattern: Pattern for the queue name"
-  [client queue-name message & options]
-  (let [{:keys [direction pattern _at _in _retry _retry-delay]
-         :or {direction :l
-              pattern :rq}
-         :as opts} options
-        packed-queue-name (utils/pack-pattern pattern queue-name)
-        encoded-message (into-array [(pr-str message)])
-        pushed-count (if (= direction :l)
-                       (.lpush @client packed-queue-name encoded-message)
-                       (.rpush @client packed-queue-name encoded-message))]
-
-    (log/debug "pushed to queue"
-               {:client client
-                :queue-name packed-queue-name
-                :message message
-                :options opts
-                :pushed-count pushed-count})
-
-    pushed-count))
-
-(defn pop!
-  "Pop a message from a queue.
-
-  Options:
-
-  - direction: Direction to pop the message (:l or :r)
-  - pattern: Pattern for the queue name"
-  [client queue-name & options]
-  (let [{:keys [direction pattern]
-         :or {direction :l
-              pattern :rq}
-         :as opts} options
-        packed-queue-name (utils/pack-pattern pattern queue-name)
-        message (if (= direction :l)
-                  (.lpop @client packed-queue-name)
-                  (.rpop @client packed-queue-name))]
-
-    (when message
-
-      (log/debug "popped from queue"
-                 {:client client
-                  :queue-name packed-queue-name
-                  :options opts
-                  :message message})
-
-      (edn/read-string message))))
-
-(defn llen
-  "Get the size of a queue.
-
-  Parameters:
-  - client: Redis client
-  - queue-name: Name of the queue
-  - options: Optional parameters, including:
-    - pattern: Pattern for the queue name"
-  [client queue-name & options]
-  (let [{:keys [pattern]
-         :or {pattern :rq}} options]
-    (.llen @client (utils/pack-pattern pattern queue-name))))
+   [com.moclojer.internal.reflection :as reflection]))
+
+(def allowlist
+  #{"lpush" "rpush" "lpop" "rpop"  "brpop"
+    "blpop" "lrange" "lindex" "lset" "lrem"
+    "llen" "linsert" "ltrim" "rpoplpush"
+    "brpoplpush"  "lmove"})
+
+(def commands
+  (for [[method parameters] (reflection/get-klazz-methods
+                             redis.clients.jedis.JedisPooled
+                             allowlist)]
+    (eval `(reflection/->wrap-method ~method ~parameters))))

src/com/moclojer/rq/utils.clj

@@ -1,4 +1,7 @@
-(ns com.moclojer.rq.utils)
+(ns com.moclojer.rq.utils
+  (:require
+   [clojure.data.json :as json]
+   [clojure.edn :as edn]))
 
 (defn- pattern->str
   "Adapts given pattern keyword to a know internal pattern. Raises
@@ -21,3 +24,56 @@
 (defn unpack-pattern
   [pattern queue-name]
   (subs queue-name (count (pattern->str pattern))))
+
+(defn- keyword-enc->fn
+  [enc]
+  (let [fns {:none identity
+             :edn pr-str
+             :json json/write-str
+             :array #(into-array (map pr-str %))
+             :edn-array #(into-array (map pr-str %))
+             :json-array #(into-array (map json/write-str %))}]
+    (or (get fns enc)
+        (throw (ex-info (str "No encoding " (name enc))
+                        {:cause :illegal-argument
+                         :value enc
+                         :expected (keys fns)})))))
+
+(defn encode
+  [enc message]
+  ((cond
+     (keyword? enc) (keyword-enc->fn enc)
+     (fn? enc) enc
+     :else (throw (ex-info
+                   (str "`encoding` must be either keyword or function")
+                   {:cause :illegal-argument
+                    :value enc
+                    :expected #{keyword? fn?}})))
+   message))
+
+(defn- keyword-dec->fn
+  [dec]
+  (let [json-dec-fn #(json/read-str % :key-fn keyword)
+        fns {:none identity
+             :edn edn/read-string
+             :json json-dec-fn
+             :array vec
+             :edn-array #(vec (map edn/read-string %))
+             :json-array #(vec (map json-dec-fn %))}]
+    (or (get fns dec)
+        (throw (ex-info (str "No decoding " (name dec))
+                        {:cause :illegal-argument
+                         :value dec
+                         :expected (keys fns)})))))
+
+(defn decode
+  [dec message]
+  ((cond
+     (keyword? dec) (keyword-dec->fn dec)
+     (fn? dec) dec
+     :else (throw (ex-info
+                   (str "`decoding` must be either keyword or function")
+                   {:cause :illegal-argument
+                    :value dec
+                    :expected #{keyword? fn?}})))
+   message))

test/com/moclojer/rq/queue_test.clj

@@ -11,17 +11,26 @@
         message (utils/gen-message)]
 
     (t/testing "raw"
-      (rq-queue/push! client queue-name message)
-      (rq-queue/push! client queue-name (utils/gen-message))
+      (rq-queue/lpush client queue-name message)
+      (rq-queue/lpush client queue-name (utils/gen-message))
       (t/is (= 2 (rq-queue/llen client queue-name)))
-      (t/is (= message (rq-queue/pop! client queue-name :direction :r))))
+      (t/is (= message (rq-queue/rpop client queue-name 1))))
 
     (t/testing "direction"
-      (rq-queue/push! client queue-name message :direction :r)
-      (t/is (= message (rq-queue/pop! client queue-name :direction :r))))
+      (rq-queue/rpush client queue-name message)
+      (t/is (= message (rq-queue/rpop client queue-name))))
 
     (t/testing "pattern"
-      (rq-queue/push! client queue-name message :pattern :pending)
-      (t/is (= message (rq-queue/pop! client queue-name :pattern :pending))))
+      (rq-queue/lpush client queue-name message :pattern :pending)
+      (t/is (= message (rq-queue/rpop client queue-name :pattern :pending))))
 
     (rq/close-client client)))
+
+(comment
+  (def my-client (rq/create-client "redis://localhost:6379"))
+
+  (.lpush @my-client)
+
+  (rq/close-client my-client)
+  ;;
+  )

test/com/moclojer/rq/utils_test.clj

@@ -1,5 +1,6 @@
 (ns com.moclojer.rq.utils-test
   (:require
+   [clojure.string :as str]
    [clojure.test :as t]
    [com.moclojer.rq.utils :as utils]))
 
@@ -13,3 +14,26 @@
    (t/is "my-queue" (utils/unpack-pattern :rq "rq:my-queue"))
    (t/is "my-queue" (utils/unpack-pattern :pubsub "rq:pubsub:my-queue"))
    (t/is "my-queue" (utils/unpack-pattern :pending "rq:pubsub:pending:my-queue"))])
+
+(t/deftest encode-test
+  (t/testing "keyword encoders"
+    [(t/is (= "hello world" (utils/encode :none "hello world")))
+     (t/is (= "{:hello? true}" (utils/encode :edn {:hello? true})))
+     (t/is (= "{\"hello?\":true}" (utils/encode :json {:hello? true})))
+     (t/is (= ["3" "true"] (vec (utils/encode :array [3 true]))))
+     (t/is (= ["{\"hello?\":true}"] (vec (utils/encode
+                                          :json-array
+                                          [{:hello? true}]))))])
+  (t/testing "function encoder"
+    (t/is (= "HELLO WORLD" (utils/encode str/upper-case "hello world")))))
+
+(t/deftest decode-test
+  (t/testing "keyword decoders"
+    [(t/is (= "hello world" (utils/decode :none "hello world")))
+     (t/is (= {:hello? true} (utils/decode :edn "{:hello? true}")))
+     (t/is (= {:hello? true} (utils/decode :json "{\"hello?\":true}")))
+     (t/is (= ["3" "true"] (utils/decode :array (into-array ["3" "true"]))))
+     (t/is (= [3 true] (utils/decode :edn-array (into-array ["3" "true"]))))
+     (t/is (= [{:hello? true}] (utils/decode
+                                :json-array
+                                (into-array ["{\"hello?\":true}"]))))]))