Merge pull request #83 from macielti/add-dlq-capabilities-to-rebbitmq…

…-consumer Add DQL capabilities to RabbitMQ consumer component
macielti · Aug 13, 2023 · e1fbe98 · e1fbe98
2 parents 0f8042d + 4cadc33
commit e1fbe98
Show file tree

Hide file tree

Showing 40 changed files with 181 additions and 51 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,14 @@ of [keepachangelog.com](http://keepachangelog.com/).
 
 ## [Unreleased]
 
+## [20.30.36] - 2023-08-12
+
+### Added
+
+- Implemented integration of RabbitMQ consumer component with Dead Letter Queue
+  service [Wraith King](https://github.com/macielti/wraith-king).
+- Added schema validation for message payloads for RabbitMQ consumer component.
+
 ## [19.30.36] - 2023-08-01
 
 ### Added
@@ -470,7 +478,9 @@ of [keepachangelog.com](http://keepachangelog.com/).
 
 - Add `loose-schema` function.
 
-[Unreleased]: https://github.com/macielti/common-clj/compare/v19.30.36...HEAD
+[Unreleased]: https://github.com/macielti/common-clj/compare/v20.30.36...HEAD
+
+[19.30.36]: https://github.com/macielti/common-clj/compare/v19.30.36...v20.30.36
 
 [19.30.36]: https://github.com/macielti/common-clj/compare/v19.29.36...v19.30.36
 

diff --git a/docs/common-clj.auth.core.html b/docs/common-clj.auth.core.html
diff --git a/docs/common-clj.component.config.html b/docs/common-clj.component.config.html
diff --git a/docs/common-clj.component.datalevin.html b/docs/common-clj.component.datalevin.html
diff --git a/docs/common-clj.component.datomic.html b/docs/common-clj.component.datomic.html
diff --git a/docs/common-clj.component.helper.core.html b/docs/common-clj.component.helper.core.html
diff --git a/docs/common-clj.component.http-client.html b/docs/common-clj.component.http-client.html
diff --git a/docs/common-clj.component.http.html b/docs/common-clj.component.http.html
diff --git a/docs/common-clj.component.kafka.adapters.html b/docs/common-clj.component.kafka.adapters.html
diff --git a/docs/common-clj.component.kafka.consumer.html b/docs/common-clj.component.kafka.consumer.html
diff --git a/docs/common-clj.component.kafka.models.html b/docs/common-clj.component.kafka.models.html
diff --git a/docs/common-clj.component.kafka.producer.html b/docs/common-clj.component.kafka.producer.html
diff --git a/docs/common-clj.component.rabbitmq.consumer.html b/docs/common-clj.component.rabbitmq.consumer.html
diff --git a/docs/common-clj.component.rabbitmq.producer.html b/docs/common-clj.component.rabbitmq.producer.html
diff --git a/docs/common-clj.component.routes.html b/docs/common-clj.component.routes.html
diff --git a/docs/common-clj.component.service.html b/docs/common-clj.component.service.html
diff --git a/docs/common-clj.component.telegram.adapters.update.html b/docs/common-clj.component.telegram.adapters.update.html
diff --git a/docs/common-clj.component.telegram.consumer.html b/docs/common-clj.component.telegram.consumer.html
diff --git a/docs/common-clj.component.telegram.models.consumer.html b/docs/common-clj.component.telegram.models.consumer.html
diff --git a/docs/common-clj.error.core.html b/docs/common-clj.error.core.html
diff --git a/docs/common-clj.io.interceptors.auth.html b/docs/common-clj.io.interceptors.auth.html
diff --git a/docs/common-clj.io.interceptors.datalevin.html b/docs/common-clj.io.interceptors.datalevin.html
diff --git a/docs/common-clj.io.interceptors.datomic.html b/docs/common-clj.io.interceptors.datomic.html
diff --git a/docs/common-clj.io.interceptors.html b/docs/common-clj.io.interceptors.html
diff --git a/docs/common-clj.keyword.core.html b/docs/common-clj.keyword.core.html
diff --git a/docs/common-clj.money.converter.html b/docs/common-clj.money.converter.html
diff --git a/docs/common-clj.schema.core.html b/docs/common-clj.schema.core.html
diff --git a/docs/common-clj.test.helper.auth.html b/docs/common-clj.test.helper.auth.html
diff --git a/docs/common-clj.test.helper.core.html b/docs/common-clj.test.helper.core.html
diff --git a/docs/common-clj.test.helper.time.html b/docs/common-clj.test.helper.time.html
diff --git a/docs/common-clj.time.core.html b/docs/common-clj.time.core.html
diff --git a/docs/common-clj.time.parser.core.html b/docs/common-clj.time.parser.core.html
diff --git a/docs/common-clj.traceability.core.html b/docs/common-clj.traceability.core.html
diff --git a/docs/index.html b/docs/index.html
diff --git a/docs/intro.html b/docs/intro.html
diff --git a/project.clj b/project.clj
@@ -1,4 +1,4 @@
-(defproject net.clojars.macielti/common-clj "19.30.36"
+(defproject net.clojars.macielti/common-clj "20.30.36"
   :description "Just common Clojure code that I use across projects"
   :url "https://github.com/macielti/common-clj"
   :license {:name "EPL-2.0 OR GPL-2.0-or-later WITH Classpath-exception-2.0"

diff --git a/resources/config_test.edn b/resources/config_test.edn
@@ -2,11 +2,13 @@
         :service-authentication {:auth-server-base-url "https://example.com"
                                  :username             "service-name"
                                  :password             "random-password"}}
- :test {:bootstrap-server       "http://localhost:9092"
-        :service                {:host "0.0.0.0"
-                                 :port 8000}
-        :topics                 ["test.example"]
-        :rabbitmq-uri           "amqp://guest:guest@localhost:5672"
-        :service-authentication {:auth-server-base-url "https://example.com"
-                                 :username             "service-name"
-                                 :password             "random-password"}}}
+ :test {:bootstrap-server                              "http://localhost:9092"
+        :service-name                                  "test-service-name"
+        :service                                       {:host "0.0.0.0"
+                                                        :port 8000}
+        :topics                                        ["test.example"]
+        :rabbitmq-uri                                  "amqp://guest:guest@localhost:5672"
+        :dead-letter-queue-service-integration-enabled true
+        :service-authentication                        {:auth-server-base-url "https://example.com"
+                                                        :username             "service-name"
+                                                        :password             "random-password"}}}
diff --git a/resources/config_test_dead_letter_disabled.edn b/resources/config_test_dead_letter_disabled.edn
@@ -0,0 +1,14 @@
+{:prod {:bootstrap-server       "http://localhost:9092"
+        :service-authentication {:auth-server-base-url "https://example.com"
+                                 :username             "service-name"
+                                 :password             "random-password"}}
+ :test {:bootstrap-server                              "http://localhost:9092"
+        :service-name                                  "test-service-name"
+        :service                                       {:host "0.0.0.0"
+                                                        :port 8000}
+        :topics                                        ["test.example"]
+        :rabbitmq-uri                                  "amqp://guest:guest@localhost:5672"
+        :dead-letter-queue-service-integration-enabled false
+        :service-authentication                        {:auth-server-base-url "https://example.com"
+                                                        :username             "service-name"
+                                                        :password             "random-password"}}}
diff --git a/src/common_clj/component/rabbitmq/consumer.clj b/src/common_clj/component/rabbitmq/consumer.clj
@@ -6,11 +6,14 @@
             [langohr.queue :as lq]
             [langohr.consumers :as lc]
             [plumbing.core :as plumbing]
-            [schema.core :as s])
+            [schema.core :as s]
+            [taoensso.timbre :as log]
+            [common-clj.component.rabbitmq.producer :as component.rabbitmq.producer])
   (:import (clojure.lang IFn)))
 
 (s/defschema Consumers
-  {s/Keyword {:handler-fn IFn}})
+  {s/Keyword {:schema     s/Any
+              :handler-fn IFn}})
 
 (defrecord Consumer [config datomic datalevin http-client rabbitmq-producer consumers]
   component/Lifecycle
@@ -24,18 +27,32 @@
                                           :rabbitmq-producer (:rabbitmq-producer rabbitmq-producer)
                                           :datomic (:datomic datomic)
                                           :datalevin (:datalevin datalevin)
-                                          :http-client (:http-client http-client))]
+                                          :http-client (:http-client http-client))
+          service-name (:service-name config-content)]
 
       (s/validate Consumers consumers)
 
       (doseq [raw-topic topics
               :let [topic (keyword raw-topic)
                     consumer (topic consumers)
-                    handler-fn (:handler-fn consumer)]]
+                    handler-fn (:handler-fn consumer)
+                    schema (:schema consumer)]]
         (lq/declare channel raw-topic {:exclusive false :auto-delete false})
         (lc/subscribe channel raw-topic
-                      (fn [_channel _meta payload] (handler-fn {:components components
-                                                                :payload    (edn/read-string (String. payload "UTF-8"))}))
+                      (fn [_channel _meta payload]
+                        (try
+                          (s/validate schema (edn/read-string (String. payload "UTF-8")))
+                          (handler-fn {:components components
+                                       :payload    (edn/read-string (String. payload "UTF-8"))})
+                          (catch Exception e
+                            (do (log/error e)
+                                (when (-> components :config :dead-letter-queue-service-integration-enabled)
+                                  (component.rabbitmq.producer/produce! {:topic   :create-dead-letter
+                                                                         :payload {:service        service-name
+                                                                                   :topic          topic
+                                                                                   :exception-info (str e)
+                                                                                   :payload        (edn/read-string (String. payload "UTF-8"))}}
+                                                                        (:rabbitmq-producer components)))))))
                       {:auto-ack true}))
 
       (merge component {:rabbitmq-consumer {:connection connection

diff --git a/test/integration/integration/rabbitmq_test.clj b/test/integration/integration/rabbitmq_test.clj
@@ -5,11 +5,16 @@
             [common-clj.component.helper.core :as component.helper]
             [common-clj.component.rabbitmq.consumer :as component.rabbitmq.consumer]
             [common-clj.component.rabbitmq.producer :as component.rabbitmq.producer]
+            [schema.core :as schema]
             [schema.test :as s]))
 
 (def test-state (atom nil))
 
-(def consumers {:test.example {:handler-fn (fn [{:keys [payload]}]
+(schema/defschema TestSchema
+  {:test schema/Keyword})
+
+(def consumers {:test.example {:schema     TestSchema
+                               :handler-fn (fn [{:keys [payload]}]
                                              (reset! test-state payload))}})
 
 (def ^:private system-test
@@ -41,3 +46,85 @@
              @(:produced-messages producer))))
 
     (component/stop system)))
+
+(s/deftest rabbitmq-consumer-and-producer-component-test-should-produce-dead-letter-on-wrong-message-schema
+  (let [system (component/start system-test)
+        producer (component.helper/get-component-content :rabbitmq-producer system)]
+
+    (testing "that we can use kafka consumer to consumer messages"
+      (component.rabbitmq.producer/produce! {:topic   :test.example
+                                             :payload {:test 1}}
+                                            producer)
+      (Thread/sleep 5000))
+
+    ;TODO: Create a helper function in order fo fetch produced messages
+    (testing "that we can retrieve produced messages"
+      (is (= [{:payload {:test 1}
+               :topic   :test.example}
+              {:payload {:exception-info "clojure.lang.ExceptionInfo: Value does not match schema: {:test (not (keyword? 1))} {:type :schema.core/error, :schema {:test Keyword}, :value {:test 1}, :error {:test (not (keyword? 1))}}"
+                         :payload        {:test 1}
+                         :service        "test-service-name"
+                         :topic          :test.example}
+               :topic   :create-dead-letter}]
+             @(:produced-messages producer))))
+
+    (component/stop system)))
+
+(def problematic-consumers {:test.example {:schema     TestSchema
+                                           :handler-fn (fn [_]
+                                                         (throw (Exception. "my exception message")))}})
+
+(def ^:private system-test-problematic-consumer-with-dead-letter-enabled
+  (component/system-map
+    :config (component.config/new-config "resources/config_test.edn" :test :edn)
+    :rabbitmq-producer (component/using (component.rabbitmq.producer/new-producer) [:config])
+    :rabbitmq-consumer (component/using (component.rabbitmq.consumer/new-consumer problematic-consumers) [:config :rabbitmq-producer])))
+
+(s/deftest rabbitmq-consumer-and-producer-component-test-should-produce-dead-letter
+  (let [system (component/start system-test-problematic-consumer-with-dead-letter-enabled)
+        producer (component.helper/get-component-content :rabbitmq-producer system)]
+
+    (testing "that we can use kafka consumer to consumer messages"
+      (component.rabbitmq.producer/produce! {:topic   :test.example
+                                             :payload {:test :ok}}
+                                            producer)
+
+      (Thread/sleep 5000))
+
+    ;TODO: Create a helper function in order fo fetch produced messages
+    (testing "that we can retrieve produced messages"
+      (is (= [{:payload {:test :ok}
+               :topic   :test.example}
+              {:payload {:exception-info "java.lang.Exception: my exception message"
+                         :payload        {:test :ok}
+                         :service        "test-service-name"
+                         :topic          :test.example}
+               :topic   :create-dead-letter}]
+             @(:produced-messages producer))))
+
+    (component/stop system)))
+
+(def ^:private system-test-problematic-consumer-with-dead-letter-disabled
+  (component/system-map
+    :config (component.config/new-config "resources/config_test_dead_letter_disabled.edn" :test :edn)
+    :rabbitmq-producer (component/using (component.rabbitmq.producer/new-producer) [:config])
+    :rabbitmq-consumer (component/using (component.rabbitmq.consumer/new-consumer problematic-consumers) [:config :rabbitmq-producer])))
+
+(s/deftest rabbitmq-consumer-and-producer-component-test-should-not-produce-dead-letter
+  (let [system (component/start system-test-problematic-consumer-with-dead-letter-disabled)
+        producer (component.helper/get-component-content :rabbitmq-producer system)]
+
+    (testing "that we can use kafka consumer to consumer messages"
+      (component.rabbitmq.producer/produce! {:topic   :test.example
+                                             :payload {:test :ok}}
+                                            producer)
+
+      (Thread/sleep 5000))
+
+    ;TODO: Create a helper function in order fo fetch produced messages
+    (testing "that we can retrieve produced messages"
+      (is (= [{:payload {:test :ok}
+               :topic   :test.example}]
+             @(:produced-messages producer))))
+
+    (component/stop system)))