processor: concurrently starts multiple pipeline loops

3 files changed, 67 insertions, 39 deletions

diff --git a/resources/config.edn b/resources/config.edn
index e5076eb..3c6ea6e 100644
--- a/resources/config.edn
+++ b/resources/config.edn
@@ -8,4 +8,5 @@
            :on-update []
            :on-delete []}
  :webhook-request-fn nil
- :feed-length 8}
+ :feed-length 8
+ :pipeline-concurrency 5}
diff --git a/src/weepinbell/processor/facade.clj b/src/weepinbell/processor/facade.clj
index fa52364..29febc2 100644
--- a/src/weepinbell/processor/facade.clj
+++ b/src/weepinbell/processor/facade.clj
@@ -16,53 +16,79 @@
   (s/keys :req-un [::on-receive ::on-create ::on-delete ::on-update]))

 (s/def ::storage (partial s/valid? :weepinbell.storage.spec/vtable))

 (s/def ::config

-  (s/keys :req-un [::base-url ::allowed-target-domains]))

+  (s/keys :req-un [::base-url ::allowed-target-domains ::pipeline-concurrency]))

 (s/def ::processor-meta

   (s/keys :req-un [::wmc ::callbacks ::storage ::config]))

 

-(defn exec-pipeline [callbacks storage config wm]

-  (let [{:keys [s t]} wm

-        id (digest (str s t))]

-    (try

-      ((:on-receive callbacks) {:webmention {:id id :source s :target t}})

-      (-> {:id id

-           :wm wm

-           :ok? true}

-          (check-storage storage)

-          verify

-          (clean storage callbacks)

-          (wrap config)

-          (write storage callbacks))

-      (catch Exception e

-        (timbre/error "unexpceted exception in pipeline: " (.getMessage e))

-        (timbre/debug e)))))

+(defn pipe

+  "Takes variable number of functions and form a pipeline.

+   Essentially a -> macro that returns a function, or a reversed comp."

+  [& functions]

+  (apply comp (reverse functions)))

+

+(defn assemble-pipeline

+  "Makes a proceesor pipeline. Returns a function that accepts webmention."

+  [{:keys [callbacks storage config]}]

+  (pipe

+   ;; initializer function

+   ;;; turns {:s ... :t ...} to pipeline map that the following functions

+   ;;; pass around; calls :on-receive hook.

+   (fn [wm]

+     (let [{:keys [s t]} wm

+           id (digest (str s t))]

+       ((:on-receive callbacks) {:webmention {:id id :source s :target t}})

+       {:id id

+        :wm wm

+        :ok? true}))

+   ;; pipeline steps

+   (partial check-storage storage)    ;; check if webmention exists

+   verify                             ;; verify if target is mentioned

+   (partial clean storage callbacks)  ;; delete old unverified webmention

+   (partial wrap config)              ;; wrap metadata into webmention

+   (partial write storage callbacks)));; write to storage

+

+(defn exec-pipeline

+  "Basically wraps (pipeline wm) call inside a try block.

+   catches exceptions and errors so they don't paralyze pipelines."

+  [pipeline wm]

+  (try

+    (pipeline wm)

+    (catch Throwable e

+      (timbre/error "unexpceted exception in pipeline: " (.getMessage e))

+      (timbre/debug e))))

+

+(defn dispatch-pipeline-loop

+  "Start a loop that takes from wmc and do pipeline when data come in."

+  [n wmc stopped pipeline]

+  (go-loop [wm (<! wmc)]

+    (let [{:keys [stop?]} wm]

+      (if (or stop? (nil? wm))

+        (do ;; handle stop signal

+          (put! stopped true)

+          (timbre/debug (str "#" n) "processor shuts down."))

+        (do ;; process found webmention

+          (timbre/info (str "#" n) "webmention found in queue, processing.")

+          (exec-pipeline pipeline wm)

+          (recur (<! wmc)))))))

 

 (defn start-processor

+  "Starts webmention processing pipelines. Returns a stopper function."

   [metadata]

   {:pre [(s/valid? ::processor-meta metadata)]

    :post [fn?]}

-  (let [{:keys [wmc callbacks storage config]} metadata

-        wmc-closed (chan)

-        close-wmc (fn []

-                    (close! wmc)

-                    (close! wmc-closed))]

-    ;; start processing loop

-    (go-loop [wm (<! wmc)]

-      (let [{:keys [stop?]} wm]

-        (if stop?

-          (do

-            (close-wmc)

-            (timbre/warn "webmention queue is cleared. processor shuts down."))

-          ;; process found webmention

-          (do

-            (timbre/info "webmention found in queue, processing.")

-            (exec-pipeline callbacks storage config wm)

-            (recur (<! wmc))))))

+  ;; wmc is webmention queue; stopped is channel where processors 

+  ;; put signals into, to notify that they're shut down

+  (let [wmc (:wmc metadata), stopped (chan)

+        concurrency (get-in metadata [:config :pipeline-concurrency])]

+    ;; dispatch pipelines

+    (doseq [n (range concurrency)

+            :let [pipeline (assemble-pipeline metadata)]]

+      (dispatch-pipeline-loop n wmc stopped pipeline))

     ;; return a function that stops the loop 

     (fn []

-      (put! wmc {:stop? true})

+      (dotimes [_ concurrency] (put! wmc {:stop? true}))

       ;; after putting the stop signal, processor might not

       ;; stop immediately. wait for it to process what's left 

       ;; inside webmention queue and close it.

-      (timbre/info "waiting for processor to shut down")

-      (<!! wmc-closed))))

+      (timbre/info "waiting for queue to clear, and processor to shut down")

+      (dotimes [_ concurrency] (<!! stopped)))))

diff --git a/test/weepinbell/processor_starter_test.clj b/test/weepinbell/processor_starter_test.clj
index f1ec273..8b604a5 100644
--- a/test/weepinbell/processor_starter_test.clj
+++ b/test/weepinbell/processor_starter_test.clj
@@ -16,7 +16,8 @@
                          :update (fn [& _])

                          :get-by-domain (fn [& _])}

                :config {:base-url "https://test.com"

-                        :allowed-target-domains ["test.com"]}}]

+                        :allowed-target-domains ["test.com"]

+                        :pipeline-concurrency 1}}]

     (start-processor pmeta)))

 

 (deftest start-processor-test