✨ Allow pluggable backends on msgbus module.

Prepare it to use different backends than redis.

backend/src/app/main.clj

@@ -58,7 +58,9 @@
     {}
 
     :app.msgbus/msgbus
-    {:uri (:redis-uri config)}
+    {:backend   (:msgbus-backend config :redis)
+     :pool      (ig/ref :app.db/pool)
+     :redis-uri (:redis-uri config)}
 
     :app.tokens/tokens
     {:sprops (ig/ref :app.setup/props)}

backend/src/app/msgbus.clj

@@ -13,6 +13,7 @@
    [app.common.exceptions :as ex]
    [app.common.spec :as us]
    [app.config :as cfg]
+   [app.db :as db]
    [app.util.blob :as blob]
    [app.util.time :as dt]
    [clojure.core.async :as a]
@@ -33,56 +34,36 @@
    io.lettuce.core.pubsub.StatefulRedisPubSubConnection
    io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands))
 
-(declare impl-publish-loop)
+(s/def ::redis-uri ::us/string)
-(declare impl-redis-pub)
-(declare impl-redis-sub)
-(declare impl-redis-unsub)
-(declare impl-subscribe-loop)
-
-
-;; --- STATE INIT: Publisher
-
-(s/def ::uri ::us/string)
 (s/def ::buffer-size ::us/integer)
 
+(defmulti init-backend :backend)
+(defmulti stop-backend :backend)
+(defmulti init-pub-loop :backend)
+(defmulti init-sub-loop :backend)
+
 (defmethod ig/pre-init-spec ::msgbus [_]
-  (s/keys :req-un [::uri]
+  (s/keys :req-un [::db/pool]
-          :opt-un [::buffer-size]))
+          :opt-un [::buffer-size ::redis-uri]))
 
 (defmethod ig/prep-key ::msgbus
   [_ cfg]
   (merge {:buffer-size 128} cfg))
 
 (defmethod ig/init-key ::msgbus
-  [_ {:keys [uri buffer-size] :as cfg}]
+  [_ {:keys [backend buffer-size] :as cfg}]
-  (let [codec    (RedisCodec/of StringCodec/UTF8 ByteArrayCodec/INSTANCE)
+  (log/debugf "initializing msgbus (backend=%s)" (name backend))
 
-        uri      (RedisURI/create uri)
+  (let [backend (init-backend cfg)
-        rclient  (RedisClient/create ^RedisURI uri)
-
-        snd-conn (.connect ^RedisClient rclient ^RedisCodec codec)
-        rcv-conn (.connectPubSub ^RedisClient rclient ^RedisCodec codec)
 
         ;; Channel used for receive publications from the application.
-        pub-chan (a/chan (a/dropping-buffer buffer-size))
+        pub-ch  (a/chan (a/dropping-buffer buffer-size))
-
-        ;; Channel used for receive data from redis
-        rcv-chan (a/chan (a/dropping-buffer buffer-size))
 
         ;; Channel used for receive subscription requests.
-        sub-chan (a/chan)
+        sub-ch  (a/chan)]
-        cch      (a/chan 1)]
 
-    (.setTimeout ^StatefulRedisConnection snd-conn ^Duration (dt/duration {:seconds 10}))
+    (init-pub-loop (assoc backend :ch pub-ch))
-    (.setTimeout ^StatefulRedisPubSubConnection rcv-conn ^Duration (dt/duration {:seconds 10}))
+    (init-sub-loop (assoc backend :ch sub-ch))
-
-    (log/debugf "initializing msgbus (uri: '%s')" (str uri))
-
-    ;; Start the sending (publishing) loop
-    (impl-publish-loop snd-conn pub-chan cch)
-
-    ;; Start the receiving (subscribing) loop
-    (impl-subscribe-loop rcv-conn rcv-chan sub-chan cch)
 
     (with-meta
       (fn run
@@ -90,85 +71,107 @@
         ([command params]
          (a/go
            (case command
-             :pub (a/>! pub-chan params)
+             :pub (a/>! pub-ch params)
-             :sub (a/>! sub-chan params)))))
+             :sub (a/>! sub-ch params)))))
 
-      {::snd-conn snd-conn
+      {::backend backend})))
-       ::rcv-conn rcv-conn
-       ::cch cch
-       ::pub-chan pub-chan
-       ::rcv-chan rcv-chan})))
 
 (defmethod ig/halt-key! ::msgbus
   [_ f]
   (let [mdata (meta f)]
-    (.close ^StatefulRedisConnection (::snd-conn mdata))
+    (stop-backend (::backend mdata))))
-    (.close ^StatefulRedisPubSubConnection (::rcv-conn mdata))
-    (a/close! (::cch mdata))
-    (a/close! (::pub-chan mdata))
-    (a/close! (::rcv-chan mdata))))
 
-(defn- impl-publish-loop
+
-  [conn pub-chan cch]
+;; --- REDIS BACKEND IMPL
-  (let [rac (.async ^StatefulRedisConnection conn)]
+
+(declare impl-redis-pub)
+(declare impl-redis-sub)
+(declare impl-redis-unsub)
+
+(defmethod init-backend :redis
+  [{:keys [redis-uri] :as cfg}]
+  (let [codec    (RedisCodec/of StringCodec/UTF8 ByteArrayCodec/INSTANCE)
+
+        uri      (RedisURI/create redis-uri)
+        rclient  (RedisClient/create ^RedisURI uri)
+
+        pub-conn (.connect ^RedisClient rclient ^RedisCodec codec)
+        sub-conn (.connectPubSub ^RedisClient rclient ^RedisCodec codec)]
+
+    (.setTimeout ^StatefulRedisConnection pub-conn ^Duration (dt/duration {:seconds 10}))
+    (.setTimeout ^StatefulRedisPubSubConnection sub-conn ^Duration (dt/duration {:seconds 10}))
+
+    (-> cfg
+        (assoc :pub-conn pub-conn)
+        (assoc :sub-conn sub-conn)
+        (assoc :close-ch (a/chan 1)))))
+
+(defmethod stop-backend :redis
+  [{:keys [pub-conn sub-conn close-ch] :as cfg}]
+  (.close ^StatefulRedisConnection pub-conn)
+  (.close ^StatefulRedisPubSubConnection sub-conn)
+  (a/close! close-ch))
+
+(defmethod init-pub-loop :redis
+  [{:keys [pub-conn ch close-ch]}]
+  (let [rac (.async ^StatefulRedisConnection pub-conn)]
     (a/go-loop []
-      (let [[val _] (a/alts! [cch pub-chan] :priority true)]
+      (let [[val _] (a/alts! [close-ch ch] :priority true)]
         (when (some? val)
           (let [result (a/<! (impl-redis-pub rac val))]
             (when (ex/exception? result)
               (log/error result "unexpected error on publish message to redis")))
           (recur))))))
 
-(defn- impl-subscribe-loop
+(defmethod init-sub-loop :redis
-  [conn rcv-chan sub-chan cch]
+  [{:keys [sub-conn ch close-ch buffer-size]}]
+  (let [rcv-ch  (a/chan (a/dropping-buffer buffer-size))
+        chans   (agent {} :error-handler #(log/error % "unexpected error on agent"))
+        tprefix (str (cfg/get :tenant) ".")
+        rac     (.async ^StatefulRedisPubSubConnection sub-conn)]
+
     ;; Add a unique listener to connection
-  (.addListener conn (reify RedisPubSubListener
+    (.addListener sub-conn
+                  (reify RedisPubSubListener
                     (message [it pattern topic message])
                     (message [it topic message]
                       ;; There are no back pressure, so we use a slidding
                       ;; buffer for cases when the pubsub broker sends
                       ;; more messages that we can process.
                       (let [val {:topic topic :message (blob/decode message)}]
-                           (when-not (a/offer! rcv-chan val)
+                        (when-not (a/offer! rcv-ch val)
                           (log/warn "dropping message on subscription loop"))))
                     (psubscribed [it pattern count])
                     (punsubscribed [it pattern count])
                     (subscribed [it topic count])
                     (unsubscribed [it topic count])))
 
-  (let [chans   (agent {} :error-handler #(log/error % "unexpected error on agent"))
+    (letfn [(subscribe-to-single-topic [nsubs topic chan]
-        tprefix (str (cfg/get :tenant) ".")
-
-        subscribe-to-single-topic
-        (fn [nsubs topic chan]
               (let [nsubs (if (nil? nsubs) #{chan} (conj nsubs chan))]
                 (when (= 1 (count nsubs))
-              (let [result (a/<!! (impl-redis-sub conn topic))]
+                  (let [result (a/<!! (impl-redis-sub rac topic))]
                     (log/tracef "opening subscription to %s" topic)
                     (when (ex/exception? result)
                       (log/errorf result "unexpected exception on subscribing to '%s'" topic))))
                 nsubs))
 
-        subscribe-to-topics
+            (subscribe-to-topics [state topics chan]
-        (fn [state topics chan]
               (let [state  (update state :chans assoc chan topics)]
                 (reduce (fn [state topic]
                           (update-in state [:topics topic] subscribe-to-single-topic topic chan))
                         state
                         topics)))
 
-        unsubscribe-from-single-topic
+            (unsubscribe-from-single-topic [nsubs topic chan]
-        (fn [nsubs topic chan]
               (let [nsubs (disj nsubs chan)]
                 (when (empty? nsubs)
-              (let [result (a/<!! (impl-redis-unsub conn topic))]
+                  (let [result (a/<!! (impl-redis-unsub rac topic))]
                     (log/tracef "closing subscription to %s" topic)
                     (when (ex/exception? result)
                       (log/errorf result "unexpected exception on unsubscribing from '%s'" topic))))
                 nsubs))
 
-        unsubscribe-channels
+            (unsubscribe-channels [state pending]
-        (fn [state pending]
               (reduce (fn [state ch]
                         (let [topics (get-in state [:chans ch])
                               state  (update state :chans dissoc ch)]
@@ -179,20 +182,20 @@
                       state
                       pending))]
 
-    ;; Asynchronous subscription loop; terminates when sub-chan is
+      ;; Asynchronous subscription loop;
-    ;; closed.
       (a/go-loop []
-      (when-let [{:keys [topics chan]} (a/<! sub-chan)]
+        (let [[val _] (a/alts! [close-ch ch])]
+          (when-let [{:keys [topics chan]} val]
             (let [topics (into #{} (map #(str tprefix %)) topics)]
               (send-off chans subscribe-to-topics topics chan)
-          (recur))))
+              (recur)))))
 
       (a/go-loop []
-      (let [[val port] (a/alts! [cch rcv-chan])]
+        (let [[val port] (a/alts! [close-ch rcv-ch])]
           (cond
             ;; Stop condition; close all underlying subscriptions and
             ;; exit. The close operation is performed asynchronously.
-          (= port cch)
+            (= port close-ch)
             (send-off chans (fn [state]
                               (log/tracef "close")
                               (->> (vals state)
@@ -202,7 +205,7 @@
 
             ;; This means we receive data from redis and we need to
             ;; forward it to the underlying subscriptions.
-          (= port rcv-chan)
+            (= port rcv-ch)
             (let [topic   (:topic val)    ; topic is already string
                   pending (loop [chans   (seq (get-in @chans [:topics topic]))
                                  pending #{}]
@@ -215,34 +218,32 @@
               (some->> (seq pending)
                        (send-off chans unsubscribe-channels))
 
-            (recur)))))))
+              (recur))))))))
 
 (defn- impl-redis-pub
-  [rac {:keys [topic message]}]
+  [^RedisAsyncCommands rac {:keys [topic message]}]
   (let [topic   (str (cfg/get :tenant) "." topic)
         message (blob/encode message)
         res     (a/chan 1)]
-    (-> (.publish ^RedisAsyncCommands rac ^String topic ^bytes message)
+    (-> (.publish rac ^String topic ^bytes message)
         (p/finally (fn [_ e]
                      (when e (a/>!! res e))
                      (a/close! res))))
     res))
 
 (defn impl-redis-sub
-  [conn topic]
+  [^RedisPubSubAsyncCommands rac topic]
-  (let [^RedisPubSubAsyncCommands cmd (.async ^StatefulRedisPubSubConnection conn)
+  (let [res (a/chan 1)]
-        res (a/chan 1)]
+    (-> (.subscribe rac (into-array String [topic]))
-    (-> (.subscribe cmd (into-array String [topic]))
         (p/finally (fn [_ e]
                      (when e (a/>!! res e))
                      (a/close! res))))
     res))
 
 (defn impl-redis-unsub
-  [conn topic]
+  [rac topic]
-  (let [^RedisPubSubAsyncCommands cmd (.async ^StatefulRedisPubSubConnection conn)
+  (let [res (a/chan 1)]
-        res (a/chan 1)]
+    (-> (.unsubscribe rac (into-array String [topic]))
-    (-> (.unsubscribe cmd (into-array String [topic]))
         (p/finally (fn [_ e]
                      (when e (a/>!! res e))
                      (a/close! res))))