lib.redis
pooled Redis clients, scripts, events, and lifecycle integration
Create RESP connection pools and compose them with optional local services, event listeners, notifications, and script state.
1 Overview
A Redis client map contains identity, host and port settings, runtime script and listener state, and an active connection pool after startup. Component wrappers coordinate supporting lifecycle steps.
2 Walkthrough
2.1 Create and start a client
(require '[lib.redis :as redis])
(def client
(redis/client-create
{:host "localhost"
:port 6379
:mode :eval}))
(def started
(redis/client:start client))2.2 Inspect and close
(redis/client-string started)
(:pool started)
@(:runtime started)
(redis/client:stop started)3 Supporting namespaces
lib.redis.event manages listeners and notification loops. lib.redis.script and lib.redis.extension support script registration and command extensions. lib.redis.bench integrates local test services.
4 API
- *default*
- client-create
- client-start
- client-steps
- client-string
- client:kill
- client:start
- client:stop
v 3.0
(defn client-create
([{:keys [id env mode host port] :as m}]
(let [id (or id (f/sid))
mode (or mode :eval)
m (merge (conn/test:config)
m)]
(merge m {:id id
:tag :redis
:mode mode
:runtime (atom {:scripts {} :listeners {}})}))))link
(r/client-create {:id "localhost" :port 17001}) => map?
v 4.0
(defn client-start
[{:keys [id host port] :as client}]
(let [pool (pool/pool {:id id
:host host
:port port
:tag "redis.pool"
:path [:redis :pool]})]
(pool/pool:start (assoc client :pool pool))))link
(with-redefs [pool/pool (constantly :pool) pool/pool:start (fn [c] (assoc c :started true))] (r/client-start {:id :id :host "h" :port 1})) => (contains {:started true :pool :pool})
v 3.0
(defn client-steps
([]
[{:key :container
:setup docker/start-runtime
:teardown docker/stop-runtime}
{:key :bench
:setup bench/bench-start
:teardown bench/bench-stop}
{:key :events
:start event/start:events-redis}
{:key :notify
:start event/start:notify-redis
:stop event/stop:notify-redis}]))link
(r/client-steps) => vector?
v 4.0
(defn client-string
[{:keys [host port pool]}]
(str "#rt.redis.client " (merge {:host host :port port}
(component/info pool))))link
(r/client-string {:host "h" :port 1 :pool :p}) => string?
NONE
(def ^{:arglists '([conn])} client:kill (component/wrap-stop pool/pool:kill (client-steps)))link
NONE
(def ^{:arglists '([conn])} client:start (component/wrap-start client-start (client-steps)))link
- +actions+
- +default+
- +lu+
- +rlu+
- ->Listener
- Listener
- action:add
- action:get
- action:list
- action:remove
- config:add
- config:get
- config:remove
- config:set
- events-parse
- events-string
- has-notify
- identity:args
- list-notify
- listener-loop
- listener-string
- listener:add
- listener:all
- listener:count
- listener:create
- listener:get
- listener:list
- listener:remove
- listener:teardown
- listener?
- map->Listener
- notify
- notify:args
- notify:wrap
- psubscribe
- psubscribe:wrap
- punsubscribe
- start:events-redis
- start:notify-redis
- stop:notify-redis
- subscribe
- subscribe:wrap
- unnotify
- unsubscribe
NONE
(impl/defimpl Listener [type id namespace handler connection]
:string listener-string)link
Listener ^
NONE
(impl/defimpl Listener [type id namespace handler connection]
:string listener-string)link
action:add ^
[key {:keys [subscribe unsubscribe args wrap], :as m}]
Added 3.0
adds an action from registry
v 3.0
(defn action:add
([key {:keys [subscribe unsubscribe args wrap] :as m}]
(swap! +actions+ assoc key m)))link
(action:add :test {:a 1}) => (contains {:test {:a 1}})
v 3.0
(defn action:get
([key]
(or (get @+actions+ key)
(throw (ex-info "Not found" {:input key})))))link
(action:add :test {:a 1}) (action:get :test) => {:a 1}
v 3.0
(defn action:remove
([key]
(swap! +actions+ dissoc key)))link
(action:add :test {:a 1}) (action:remove :test) => map?
v 3.0
(defn config:add
([redis events]
(config:add redis events nil))
([redis events opts]
(do (config:set redis (apply conj (config:get redis opts) events) opts)
(config:get redis opts))))link
(with-redefs [config:get (constantly #{:string}) config:set (constantly "OK")] (config:add |client| #{:generic})) => #{:string}
v 3.0
(defn config:get
([redis]
(config:get redis nil))
([redis opts]
((comp events-parse f/strn second)
(pool/pool:request-single redis ["CONFIG" "GET" "notify-keyspace-events"] opts))))link
(with-redefs [pool/pool:request-single (constantly ["key" "K$g"])] (config:get |client|)) => #{:string :generic}
v 3.0
(defn config:remove
([redis events]
(config:remove redis events nil))
([redis events opts]
(let [current (config:get redis)]
(config:set redis (clojure.set/difference current (set events)))
(config:get redis opts))))link
(with-redefs [config:get (constantly #{:string :generic}) config:set (constantly "OK")] (config:remove |client| #{:generic})) => #{:string :generic}
v 3.0
(defn config:set
([redis events]
(config:set redis events nil))
([redis events opts]
(let [s (events-string events)
opts (cc/req:opts opts)]
(pool/pool:request-single redis ["CONFIG" "SET" "notify-keyspace-events" (str "K" s)]
opts))))link
(with-redefs [pool/pool:request-single (constantly "OK")] (config:set |client| #{:string})) => "OK"
v 3.0
(defn events-parse
([^String s]
(set (keep +rlu+ s))))link
(events-parse "h$g") => #{:hash :string :generic}
v 3.0
(defn events-string
([events]
(apply str (keep +lu+ events))))link
(events-string #{:string :hash :generic}) => "h$g"
v 3.0
(defn has-notify
([redis id]
(boolean (get-in @(:runtime redis) [:listeners :notify id]))))link
(swap! (:runtime |client|) assoc-in [:listeners :notify :id] {}) (has-notify |client| :id) => true
v 3.0
(defn list-notify
([redis]
(collection/map-vals #(select-keys % [:pattern :handler])
(get-in @(:runtime redis) [:listeners :notify]))))link
(list-notify |client|) => {:id {}}
v 3.0
(defn listener-loop
([conn f]
(loop []
(let [msg (wire/read conn)]
(f msg)
(recur)))))link
(let [seen (atom []) count (atom 0)] (try (with-redefs [wire/read (fn [_] (let [n (swap! count inc)] (if (< n 3) n (throw (ex-info "done" {})))))] (listener-loop :conn #(swap! seen conj %))) (catch clojure.lang.ExceptionInfo _)) @seen) => [1 2]
listener-string ^
[{:keys [type connection], :as listener}]
Added 3.0
string description of a listener
v 3.0
(defn listener-string
([{:keys [type connection] :as listener}]
(let [{:keys [id args raw thread]} connection]
(str "#" (name type) (:id listener) " "
{:id id :args args
:running (and (not (future/future:complete? thread))
(component/started? raw))}))))link
(with-redefs [future/future:complete? (constantly false) component/started? (constantly true)] (listener-string {:type :test :id :listener :connection {:id :conn :args ["in"] :raw :raw :thread :thread}})) => "#test:listener {:id :conn, :args ["in"], :running true}"
listener:add ^
[type redis id input handler] [type {:keys [namespace runtime pool], :as redis} id input handler opts]
Added 3.0
adds a listener to the redis client
v 3.0
(defn listener:add
([type redis id input handler]
(listener:add type redis id input handler {}))
([type {:keys [namespace runtime pool] :as redis} id input handler opts]
(let [{:keys [listeners]} @runtime
id (or id (keyword (f/sid)))]
(when-not (get-in listeners [type id])
(doto (-> (listener:create redis type id input handler)
(merge opts))
(->> (swap! runtime assoc-in [:listeners type id])))))))link
(with-redefs [listener:create (constantly {:id :id :type :test}) f/sid (constantly "sid")] (listener:add :test |client| :id "in" identity)) => map?
v 3.0
(defn listener:all
([{:keys [runtime]}]
(mapcat vals (vals (:listeners @runtime)))))link
(swap! (:runtime |client|) assoc :listeners {:test {:id {:a 1}}}) (listener:all |client|) => [{:a 1}]
v 3.0
(defn listener:count
([{:keys [runtime]}]
(collection/map-vals count (:listeners @runtime))))link
(listener:count |client|) => {:test 1}
listener:create ^
[{:keys [namespace pool], :as redis} type id input handler]
Added 3.0
creates a listener
v 3.0
(defn listener:create
([{:keys [namespace pool] :as redis} type id input handler]
(let [[raw-id raw] (cc/pool:acquire pool)
action (action:get type)
args ((action :args) namespace input)
_ (wire/call raw (into [(action :subscribe)] args))
connection {:id raw-id
:args args
:pool pool
:thread (future/future
(->> ((action :wrap) handler redis)
(listener-loop raw)))
:raw raw}]
(-> {:type type :id id
:namespace namespace
:input input
:handler handler
:connection connection}
(map->Listener)))))link
(with-redefs [cc/pool:acquire (constantly [:raw-id :raw]) action:get (constantly {:args (constantly []) :subscribe "SUB" :wrap (constantly identity)}) wire/call (constantly nil) listener-loop (constantly nil)] (listener:create |client| :test :id "in" identity)) => listener?
v 3.0
(defn listener:get
([{:keys [runtime]} type id]
(get-in (:listeners @runtime)
[type id])))link
(listener:get |client| :test :id) => {:a 1}
v 3.0
(defn listener:list
([{:keys [runtime]}]
(collection/map-vals keys (:listeners @runtime))))link
(listener:list |client|) => {:test [:id]}
listener:remove ^
[type {:keys [runtime], :as redis} id]
Added 3.0
removes a listener from the client
v 3.0
(defn listener:remove
([type {:keys [runtime] :as redis} id]
(when-let [listener (-> (:listeners @runtime)
(get-in [type id]))]
(listener:teardown listener)
(swap! runtime update :listeners collection/dissoc-nested [type id])
listener)))link
(with-redefs [listener:teardown (constantly nil)] (swap! (:runtime |client|) assoc-in [:listeners :test :id] {}) (listener:remove :test |client| :id)) => {}
v 3.0
(defn listener:teardown
([{:keys [type connection] :as listener}]
(let [{:keys [id args pool thread raw]} connection
action (action:get type)
_ (wire/write raw (into [(action :unsubscribe) args]))
_ (future/future:cancel thread)
_ (cc/pool:release pool id true)]
id)))link
(with-redefs [action:get (constantly {:unsubscribe "UNSUB"}) wire/write (constantly nil) future/future:cancel (constantly nil) cc/pool:release (constantly nil)] (listener:teardown {:type :test :connection {:id :id :args [] :pool :pool :thread :thread :raw :raw}})) => :id
v 3.0
(defn listener?
([obj]
(instance? Listener obj)))link
(listener? (map->Listener {:type :test :id :id :namespace "ns" :handler identity :connection {}})) => true
NONE
(impl/defimpl Listener [type id namespace handler connection]
:string listener-string)link
v 3.0
(defn notify
([redis id pattern handler]
(listener:add :notify redis id pattern handler {:pattern pattern})))link
(with-redefs [listener:add (constantly :added)] (notify |client| :id "*" identity)) => :added
v 3.0
(defn notify:args
([namespace inputs]
(mapv #(str "__keyspace@*:" (common/make-key namespace %))
(collection/seqify inputs))))link
(notify:args "test" ["input"]) => ["__keyspace@*:test:input"]
notify:wrap ^
[handler {:keys [namespace], :as redis}]
Added 3.0
wrapper for the notify delivery function
v 3.0
(defn notify:wrap
([handler {:keys [namespace] :as redis}]
(fn [msg]
(let [[type _ ^String path cmd :as msg] (wire/coerce msg :string)]
(cond (= type "pmessage")
(let [i (.indexOf path ":")
key (subs path (inc i))]
(env/explode
(handler redis (common/unmake-key namespace key)))))))))link
((notify:wrap (fn [redis key] key) {:namespace "test"}) [(.getBytes "pmessage") (.getBytes "pattern") (.getBytes "__keyspace@*:test:input") (.getBytes "set")]) => "input"
v 3.0
(defn psubscribe
([redis id pattern handler]
(listener:add :psubscribe redis id pattern handler)))link
(with-redefs [listener:add (constantly :added)] (psubscribe |client| :id ["*"] identity)) => :added
psubscribe:wrap ^
[handler {:keys [format], :as redis}]
Added 3.0
wrapper for the psubscribe delivery function
v 3.0
(defn psubscribe:wrap
([handler {:keys [format] :as redis}]
(fn [msg]
(let [[type pattern channel data] (seq msg)]
(if (= (wire/coerce type :string) "pmessage")
(handler (wire/coerce channel :string)
redis
nil
(wire/coerce data format)))))))link
((psubscribe:wrap (fn [ch redis _ msg] msg) {:format :edn}) [(.getBytes "pmessage") (.getBytes "pattern") (.getBytes "channel") (.getBytes "{:a 1}")]) => {:a 1}
v 3.0
(defn punsubscribe
([redis id]
(listener:remove :psubscribe redis id)))link
(with-redefs [listener:remove (constantly :removed)] (punsubscribe |client| :id)) => :removed
v 3.0
(defn start:events-redis
([redis events]
(config:set redis events)
redis))link
(with-redefs [config:set (constantly nil)] (start:events-redis |client| #{})) => map?
v 3.0
(defn start:notify-redis
([redis listeners]
(doseq [[id {:keys [pattern handler]}] listeners]
(notify redis id pattern handler))
redis))link
(with-redefs [notify (constantly nil)] (start:notify-redis |client| {:id {:pattern "*" :handler identity}})) => map?
v 3.0
(defn stop:notify-redis
([redis listeners]
(doseq [[id _] listeners]
(unnotify redis id))
redis))link
(with-redefs [unnotify (constantly nil)] (stop:notify-redis |client| {:id {}})) => map?
v 3.0
(defn subscribe
([redis id channels handler]
(listener:add :subscribe redis id channels handler)))link
(with-redefs [listener:add (constantly :added)] (subscribe |client| :id ["ch"] identity)) => :added
subscribe:wrap ^
[handler {:keys [format], :as redis}]
Added 3.0
wrapper for the subscribe delivery function
v 3.0
(defn subscribe:wrap
([handler {:keys [format] :as redis}]
(fn [msg]
(let [[type channel data] (seq msg)]
(if (= (wire/coerce type :string) "message")
(handler (wire/coerce channel :string)
redis
nil
(wire/coerce data format)))))))link
((subscribe:wrap (fn [ch redis _ msg] msg) {:format :edn}) [(.getBytes "message") (.getBytes "channel") (.getBytes "{:a 1}")]) => {:a 1}
+additional+ ^
NONE
(def +additional+
(gen/select-commands {:group [:pubsub :scripting :hyperloglog :transactions]
:custom {:pfadd {:arguments {1 {:type :data}}}
:publish {:arguments {1 {:type :data}}}
:psubscribe {:arguments [{:type :key :sym 'patterns}]}
:subscribe {:arguments [{:type :key}]}}}))link
+data+ ^
NONE
(def +data+
(gen/select-commands {:group [:list :hash :set]
:custom {:lpos {:arguments {1 {:type :data}}}
:lpush {:arguments {1 {:type :data}}}
:lpushx {:arguments {1 {:type :data}}}
:lrem {:arguments {2 {:type :data}}}
:lset {:arguments {2 {:type :data}}}
:rpush {:arguments {1 {:type :data}}}
:rpushx {:arguments {1 {:type :data}}}
:hgetall {:return `common/return:kv-hash}
:hkeys {:return `common/return:string}
:hmset {:arguments {1 {:sym 'args :process `common/process:kv-hash}}}
:hsetnx {:arguments {2 {:type :data}}}
:sadd {:arguments {1 {:type :data}}}
:srem {:arguments {1 {:type :data}}}
:sismember {:arguments {1 {:type :data}}}
:smove {:arguments {2 {:type :data}}}}
:replace {:hset {:id :hset
:return :ack
:arguments [{:type :key}
{:type :string :name "field"}
{:type :data}]}}}))link
+generic+ ^
NONE
(def +generic+
(gen/select-commands {:group :generic
:exclude [:sort :migrate]
:custom {:keys {:arguments [{:name "pattern", :type :key}]
:return `common/return:keys}}}))link
+management+ ^
NONE
(def +management+
(gen/select-commands {:group [:server :cluster :connection]
:exclude [:hello]
:custom {:acl-log {:arguments [{:name "count"}]}
:shutdown {:arguments [{:name "mode"}]}}}))link
+string+ ^
NONE
(def +string+
(gen/select-commands {:group :string
:exclude [:bitcount :bitfield]
:custom {:append {:arguments {1 {:type :data}}}
:getset {:arguments {1 {:type :data}}}
:psetex {:arguments {2 {:type :data}}}
:setex {:arguments {2 {:type :data}}}
:setnx {:arguments {1 {:type :data}}}
:msetnx {:arguments [{:sym 'args :process `common/process:kv}]}
:mset {:arguments [{:sym 'args :process `common/process:kv}]}}
:replace {:set {:id :set
:return :ack
:arguments [{:type :key}
{:type :data}
{:type :optional
:sym 'optional
:display '{:keys [expiry unit mode] :as optional}
:process `optional:set}]}}}))link
v 4.0
(defn optional:set
([{:keys [expiry unit mode]} _]
(cond-> []
mode (conj (case mode
:not-exist "NX"
:exists "XX"))
expiry (concat (cond (= expiry :keep)
["KEEPTTL"]
(= unit :ms)
["PX" expiry]
:else
["EX" expiry]))
:then vec)))link
(optional:set {:expiry 10 :unit :ms :mode :exists} nil) => ["XX" "PX" 10]