Adding async interceptor helpers (#1)

majorcluster · Mar 31, 2023 · d45187b · d45187b
1 parent 03a732a
commit d45187b
Show file tree

Hide file tree

Showing 10 changed files with 274 additions and 16 deletions.
diff --git a/README.md b/README.md
@@ -16,7 +16,7 @@ A Clojure library designed to extend usual pedestal api setup, providing:
 ```
 
 ## Examples:
-[Read the docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/index.md)
+[Read the docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/index.md)
 
 ## Publish
 ### Requirements
@@ -29,4 +29,4 @@ export GPG_TTY=$(tty) && lein deploy clojars
 ```
 
 ## Documentation
-[Link](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/index.md)
+[Link](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/index.md)
diff --git a/doc/async_interceptors.md b/doc/async_interceptors.md
@@ -0,0 +1,49 @@
+## Index
+- Functions
+  - [async-blocker-interceptor](#async-blocker-interceptor)
+  - [async-fetch-output-interceptor](#async-fetch-output-interceptor)
+  - [async-output-interceptor](#async-output-interceptor)
+
+## Functions
+
+- <h3><a id='async-blocker-interceptor'></a><span style="color:green">async-blocker-interceptor</span> [] [merge-data-fn]</h3><br>
+  Async blocker interceptor, the one mandatory to resolve async-channels created by `async-fetch-output-interceptor`
+  before jumping into the handlers. Optional merge-data-fn is the function used to merge different data returned by the interceptors.
+  The default fn is `merge`. <br>
+  <br>
+  - merge-data-fn ^fn : optional merge function, when not sent, `merge` is used as default<br>
+  - returns ^map context map <br>
+```clojure
+    (async-blocker-interceptor)
+```
+```clojure
+    (async-blocker-interceptor my-merge-fn)
+``` 
+
+- <h3><a id='async-fetch-output-interceptor'></a><span style="color:green">async-fetch-output-interceptor</span> [m]</h3><br>
+  Creates an async interceptor that receives a `{:name string? :enter fn? :leave fn?}` as param.<br>
+  It creates a channel and adds it the context `:async-channels` map with interceptor `:name` as key.<br>
+  The optional `:enter` function is triggered and the result is executed async and put to the channel while the queue goes on.<br>
+  The optional `:leave` function is triggered async, but no result is stored anywhere <br>
+  <br>
+  - m ^map : interceptor map `{:name string? :enter fn? :leave fn?}` one of enter or leave is mandatory, both are also accepted<br>
+  - returns ^map context map <br>
+```clojure
+    (async-fetch-output-interceptor {:name :async-i-name
+                                     :enter (fn [context] 
+                                              ;... my body sync or async to be paralellized
+                                              )})
+```
+
+- <h3><a id='async-output-interceptor'></a><span style="color:green">async-output-interceptor</span> [m]</h3><br>
+  Creates an async interceptor that receives a `{:name string? :enter fn?}` as param.<br>
+  It executes the `:enter` and/or `:leave` function async and the queue goes on while it is executed by the async threads in parallel. <br>
+  <br>
+  - m ^map : interceptor map `{:name string? :enter fn? :leave fn?}` one of enter or leave is mandatory, both are also accepted<br>
+  - returns ^map context map <br>
+```clojure
+    (async-output-interceptor {:name :async-i-name
+                               :enter (fn [context] 
+                                           ;... my body sync or async to be parallelized
+                                           )})
+```
diff --git a/doc/index.md b/doc/index.md
@@ -1,6 +1,7 @@
 # pedestal-api-helper
 Documentation:
 
-- [interceptors](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/interceptors.md)
-- [params-helper](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)
-- [validation](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/validation.md)
+- [async-interceptors](async_interceptors.md)
+- [interceptors](interceptors.md)
+- [params-helper](params_helper.md)
+- [validation](validation.md)
diff --git a/integration/clj/pedestal_api_helper/async_interceptors_i_test.clj b/integration/clj/pedestal_api_helper/async_interceptors_i_test.clj
@@ -0,0 +1,69 @@
+(ns clj.pedestal-api-helper.async-interceptors-i-test
+  (:require [clj.pedestal-api-helper.core-test :refer [service-map]]
+            [clojure.edn :as edn]
+            [clojure.test :refer :all]
+            [io.pedestal.http :as http]
+            [io.pedestal.test :refer [response-for]]
+            [mockfn.macros :refer [verifying]]
+            [mockfn.matchers :refer [exactly]]
+            [pedestal-api-helper.async-interceptors :as async-i]))
+
+(defn get-foo
+  [request]
+  {:status 200 :body (-> request :async-data)})
+
+(defn http-out
+  [v] v)
+
+(def fetch-interceptors
+  [(async-i/async-fetch-output-interceptor {:name :async-1
+                                            :enter (fn [_] (http-out 1) 1)})
+   (async-i/async-blocker-interceptor)])
+
+(def output-enter-interceptors
+  [(async-i/async-output-interceptor {:name :async-2
+                                      :enter (fn [_] (http-out 2) 2)})])
+
+(def output-leave-interceptors
+  [(async-i/async-output-interceptor {:name :async-3
+                                      :leave (fn [_] (http-out 3) 3)})])
+
+(def all-interceptors
+  (into [] (concat output-enter-interceptors fetch-interceptors output-leave-interceptors)))
+
+(def service
+  (::http/service-fn (http/create-servlet (assoc service-map
+                                            ::http/routes #{["/foo" :get (conj fetch-interceptors
+                                                                               `get-foo) :route-name :get-foo]
+                                                            ["/foo-2" :get (conj output-enter-interceptors
+                                                                                 `get-foo) :route-name :get-foo-2]
+                                                            ["/foo-3" :get (conj output-leave-interceptors
+                                                                                 `get-foo) :route-name :get-foo-3]
+                                                            ["/foo-4" :get (conj all-interceptors
+                                                                                 `get-foo) :route-name :get-foo-4]}))))
+
+(deftest foo-get-test
+  (verifying [(http-out 1) 1 (exactly 1)]
+    (let [response (response-for service :get "/foo")]
+      (is (= (:status response) 200))
+      (is (= (edn/read-string (:body response)) {:async-1 1})))))
+
+(deftest foo-2-get-test
+  (verifying [(http-out 2) 2 (exactly 1)]
+    (let [response (response-for service :get "/foo-2")]
+      (is (= (:status response) 200))
+      (is (= (:body response) "")))))
+
+(deftest foo-3-get-test
+  (verifying [(http-out 3) 3 (exactly 1)]
+    (let [response (response-for service :get "/foo-3")]
+      (is (= (:status response) 200))
+      (is (= (:body response) "")))))
+
+(deftest foo-4-get-test
+  (verifying [(http-out 2) 2 (exactly 1)
+              (http-out 1) 1 (exactly 1)
+              (http-out 3) 3 (exactly 1)]
+    (let [response (response-for service :get "/foo-4")]
+      (is (= (:status response) 200))
+      (is (= (edn/read-string (:body response)) {:async-1 1})))))
diff --git a/integration/clj/pedestal_api_helper/core_test.clj b/integration/clj/pedestal_api_helper/core_test.clj
@@ -0,0 +1,13 @@
+(ns clj.pedestal-api-helper.core-test
+  (:require [clojure.test :refer :all]
+            [io.pedestal.http :as http]))
+
+(def service-map {:env :test
+                  ::http/routes #{}
+                  ::http/resource-path "/public"
+
+                  ::http/type :jetty
+                  ::http/port 8080
+                  ::http/container-options {:h2c? true
+                                            :h2? false
+                                            :ssl? false}})
diff --git a/project.clj b/project.clj
@@ -1,13 +1,20 @@
-(defproject org.clojars.majorcluster/pedestal-api-helper "0.8.1"
+(defproject org.clojars.majorcluster/pedestal-api-helper "0.9.0"
   :description "Useful simple tools to be used for pedestal APIs"
   :url "https://github.com/majorcluster/pedestal-api-helper"
   :license {:name "The MIT License"
             :url "http://opensource.org/licenses/MIT"}
   :dependencies [[org.clojure/clojure "1.11.1"]
-                 [org.clojure/data.json "2.4.0"]]
+                 [org.clojure/data.json "2.4.0"]
+                 [org.clojure/core.async "1.6.673"]]
   :source-paths ["src/clj"]
+  :test-paths ["test","integration"]
   :profiles {:dev {:plugins [[com.github.clojure-lsp/lein-clojure-lsp "1.3.17"]]
-                   :dependencies [[nubank/matcher-combinators "3.8.3"]]}}
+                   :dependencies [[nubank/matcher-combinators "3.8.3"]
+                                  [io.pedestal/pedestal.service "0.5.10"]
+                                  [io.pedestal/pedestal.route "0.5.10"]
+                                  [io.pedestal/pedestal.jetty "0.5.10"]
+                                  [org.slf4j/slf4j-simple "1.7.28"]
+                                  [nubank/mockfn "0.7.0"]]}}
   :aliases {"diagnostics"     ["clojure-lsp" "diagnostics"]
             "format"          ["clojure-lsp" "format" "--dry"]
             "format-fix"      ["clojure-lsp" "format"]

diff --git a/src/clj/pedestal_api_helper/async_interceptors.clj b/src/clj/pedestal_api_helper/async_interceptors.clj
@@ -0,0 +1,64 @@
+(ns pedestal-api-helper.async-interceptors
+  (:require [clojure.core.async :as async]))
+
+(defn- push-async-channel
+  [context channel-id channel]
+  (assoc-in context [:request :async-channels channel-id] channel))
+
+(defn- push-async-data
+  ([context data]
+   (assoc-in context [:request :async-data] (conj (get-in context [:request :async-data] [])
+                                                  data)))
+  ([context id data]
+   (assoc-in context [:request :async-data id] data)))
+
+(defn- executes-context-async-fn
+  [m-fn context]
+  (async/go
+    (do
+      (println "exit executes-context-async-fn")
+      (m-fn context)))
+  context)
+
+(defn async-fetch-output-interceptor
+  "Creates an async interceptor that receives a `{:name string? :enter fn? :leave fn?}` as param.
+   It creates a channel and adds it the context `:async-channels` map with interceptor `:name` as key.
+   The optional `:enter` function is triggered and the result is executed async and put to the channel while the queue goes on.
+   The optional `:leave` function is triggered async, but no result is stored anywhere"
+  [m]
+  (let [enter-m (if (:enter m) (assoc m :enter (fn [context]
+                                                 (let [chan (async/chan)]
+                                                   (async/go
+                                                     (async/>! chan (get-in (push-async-data context (:name m) ((:enter m) context)) [:request :async-data])))
+                                                   (push-async-channel context (:name m) chan))))
+                    {})
+        leave-m (if (:leave m) (assoc m :leave #(executes-context-async-fn (:leave m) %))
+                    {})]
+    (merge enter-m leave-m)))
+
+(defn async-output-interceptor
+  "Creates an async interceptor that receives a `{:name string? :enter fn?}` as param.
+   It executes the `:enter` and/or `:leave` function async and the queue goes on while it is executed by the async threads in parallel."
+  [m]
+  (let [enter-m (if (:enter m) (assoc m :enter #(executes-context-async-fn (:enter m) %))
+                    {})
+        leave-m (if (:leave m) (assoc m :leave #(executes-context-async-fn (:leave m) %))
+                    {})]
+    (merge enter-m leave-m)))
+
+(defn async-blocker-interceptor
+  "Async blocker interceptor, the one mandatory to resolve async-channels created by all `async-fetch-output-interceptor`
+   before jumping into the handlers. Optional merge-data-fn is the function used to merge different data returned by the interceptors.
+   The default fn is `merge`"
+  ([merge-data-fn]
+   {:name  ::async-blocker
+    :enter (fn [context]
+             (println "async-blocker-interceptor")
+             (let [channels (-> context :request :async-channels vals)
+                   merged-channels (async/map merge-data-fn channels)
+                   async-data (async/<!! merged-channels)
+                   context (assoc-in context [:request :async-data] async-data)]
+               (async/close! merged-channels)
+               context))})
+  ([]
+   (async-blocker-interceptor merge)))
diff --git a/src/clj/pedestal_api_helper/params_helper.clj b/src/clj/pedestal_api_helper/params_helper.clj
@@ -6,23 +6,23 @@
   #"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
 
 (defn uuid
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   []
   (UUID/randomUUID))
 
 (defn uuid-as-string
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   [uuid]
   (.toString uuid))
 
 (defn is-uuid
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   [id]
   (cond (string? id) (re-matches uuid-pattern id)
         :else false))
 
 (defn validate-mandatory
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   ([body fields message-untranslated]
    (let [fields (map keyword fields)
          not-present (filter (fn [field]
@@ -38,15 +38,15 @@
    (validate-mandatory body fields "Field %s is not present")))
 
 (defn extract-field-value
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   [field body]
   (let [value (field body)
         is-uuid (is-uuid value)]
     (cond is-uuid (UUID/fromString value)
           :else value)))
 
 (defn mop-fields
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   ([body fields opts]
    (let [fields (map keyword fields)
          ignore-uuid (get opts :ignore-uuid false)]
@@ -62,7 +62,7 @@
    (mop-fields body fields {})))
 
 (defn validate-and-mop!!
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/params_helper.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/params_helper.md)"
   ([body
     to-validate
     accepted

diff --git a/src/clj/pedestal_api_helper/validation.clj b/src/clj/pedestal_api_helper/validation.clj
@@ -107,7 +107,7 @@
         :else result))
 
 (defn validate
-  "[docs](https://github.com/mtsbarbosa/pedestal-api-helper/tree/main/doc/validation.md)"
+  "[docs](https://github.com/majorcluster/pedestal-api-helper/tree/main/doc/validation.md)"
   [body fields]
   (let [body (cond (nil? body) {}
                    :else body)

diff --git a/test/clj/pedestal_api_helper/async_interceptors_test.clj b/test/clj/pedestal_api_helper/async_interceptors_test.clj
@@ -0,0 +1,55 @@
+(ns clj.pedestal-api-helper.async-interceptors-test
+  (:require [clojure.core.async :as async]
+            [clojure.test :refer :all]
+            [pedestal-api-helper.async-interceptors :as a-int])
+  (:import [clojure.core.async.impl.channels ManyToManyChannel]))
+
+(deftest async-fetch-output-interceptor-test
+  (testing "only enter and name are in the context"
+    (let [async-i (a-int/async-fetch-output-interceptor {:name :async-1
+                                                         :enter (fn [_] 1)})]
+      (is (= 2 (count (select-keys async-i [:name :enter :leave]))))))
+  (testing "only leave and name are in the context"
+    (let [async-i (a-int/async-fetch-output-interceptor {:name :async-1
+                                                         :leave (fn [_] 1)})]
+      (is (= 2 (count (select-keys async-i [:name :enter :leave]))))))
+  (testing "enter, leave and name are in the context"
+    (let [async-i (a-int/async-fetch-output-interceptor {:name :async-1
+                                                         :enter (fn [_] 1)
+                                                         :leave (fn [_] 1)})]
+      (is (= 3 (count (select-keys async-i [:name :enter :leave]))))))
+  (testing "async-channel is attached"
+    (let [async-i (a-int/async-fetch-output-interceptor {:name :async-1
+                                                         :enter (fn [_] 1)})
+          context-executed ((:enter async-i) {})]
+      (is (instance? ManyToManyChannel (-> context-executed :request :async-channels :async-1)))
+      (is (= (-> context-executed :request :async-channels :async-1 async/<!! :async-1)
+             1))))
+  (testing "async-channels are attached"
+    (let [async-i (a-int/async-fetch-output-interceptor {:name :async-1
+                                                         :enter (fn [_] 1)})
+          async-i-2 (a-int/async-fetch-output-interceptor {:name :async-2
+                                                           :enter (fn [_] 2)})
+          context-executed (-> {}
+                               ((:enter async-i))
+                               ((:enter async-i-2)))]
+      (is (instance? ManyToManyChannel (-> context-executed :request :async-channels :async-1)))
+      (is (= (-> context-executed :request :async-channels :async-1 async/<!! :async-1)
+             1))
+      (is (= (-> context-executed :request :async-channels :async-2 async/<!! :async-2)
+             2)))))
+
+(deftest async-fetch-interceptor-test
+  (testing "only enter and name are in the context"
+    (let [async-i (a-int/async-output-interceptor {:name :async-1
+                                                   :enter (fn [_] 1)})]
+      (is (= 2 (count (select-keys async-i [:name :enter :leave]))))))
+  (testing "only leave and name are in the context"
+    (let [async-i (a-int/async-output-interceptor {:name :async-1
+                                                   :leave (fn [_] 1)})]
+      (is (= 2 (count (select-keys async-i [:name :enter :leave]))))))
+  (testing "enter, leave and name are in the context"
+    (let [async-i (a-int/async-output-interceptor {:name :async-1
+                                                   :enter (fn [_] 1)
+                                                   :leave (fn [_] 1)})]
+      (is (= 3 (count (select-keys async-i [:name :enter :leave])))))))