processor, storage: clear queues before shutting down proceesor and storage worker

3 files changed, 62 insertions, 39 deletions

diff --git a/src/weepinbell/core.clj b/src/weepinbell/core.clj
index 1cc233e..2a16335 100644
--- a/src/weepinbell/core.clj
+++ b/src/weepinbell/core.clj
@@ -44,10 +44,10 @@
                         {:wmc wmc
                          :store async-store
                          :config config})]
-    (fn [] 
-      (stop-processor) 
-      (stop-worker)
-      (stop-webserver) 
+    (fn []
+      (stop-webserver) ;; stop receiving webmentions
+      (stop-processor) ;; stop processing webmentions
+      (stop-worker)    ;; stop writing to storage
       true)))
 
 ; Entry point; Terminal user interface
diff --git a/src/weepinbell/processor/facade.clj b/src/weepinbell/processor/facade.clj
index 224c739..fa52364 100644
--- a/src/weepinbell/processor/facade.clj
+++ b/src/weepinbell/processor/facade.clj
@@ -1,6 +1,6 @@
 (ns weepinbell.processor.facade

   (:require [clojure.spec.alpha :as s]

-            [clojure.core.async :refer [go-loop <! put!]]

+            [clojure.core.async :refer [go-loop <! put! close! chan <!!]]

             [taoensso.timbre :as timbre]

             [weepinbell.utils.digest :refer [digest]]

             [weepinbell.storage.spec]

@@ -20,33 +20,49 @@
 (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 start-processor

   [metadata]

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

    :post [fn?]}

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

+  (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 [s t stop?]} wm

-            id (digest (str s t))]

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

         (if stop?

-          (timbre/warn "processor stopped on stop signal.")

+          (do

+            (close-wmc)

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

           ;; process found webmention

           (do

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

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

-            (try

-              (-> {: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)))

+            (exec-pipeline callbacks storage config wm)

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

     ;; return a function that stops the loop 

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

+    (fn []

+      (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))))

diff --git a/src/weepinbell/storage/worker.clj b/src/weepinbell/storage/worker.clj
index 41d4912..1ce3675 100644
--- a/src/weepinbell/storage/worker.clj
+++ b/src/weepinbell/storage/worker.clj
@@ -1,7 +1,7 @@
 (ns weepinbell.storage.worker

   "Storage worker is a background loop that handles write operations 

    to the storage, which makes write/delete/update thread-safe."

-  (:require [clojure.core.async :refer [go-loop chan put! <! close!]]

+  (:require [clojure.core.async :refer [go-loop chan put! <! <!! close!]]

             [taoensso.timbre :as timbre]))

 

 (defn- start-worker

@@ -9,21 +9,28 @@
    and writes to storage when operation is received.

    Returns a function that stops the worker gracefully."

   [storage writer-chan]

-  (go-loop [operation (<! writer-chan)]

-    (let [{:keys [op args]} operation]

-      (if (not= op :stop)

-        (do ;; perform operation and recur

-          (try

-            (apply (op storage) args)

-            (catch Throwable e

-              (timbre/error "Something unexpected occrus while writing to storage.")

-              (timbre/error e)))

-          (recur (<! writer-chan)))

-        (do

-          (close! writer-chan)

-          (timbre/info "shutting down storage worker.")))))

-  ;; return a function that stops the loop

-  (fn [] (put! writer-chan {:op :stop})))

+  (let [queue-closed? (chan)

+        close-queue (fn []

+                      (close! writer-chan)

+                      (close! queue-closed?))]

+    (go-loop [operation (<! writer-chan)]

+      (let [{:keys [op args]} operation]

+        (if (not= op :stop)

+          (do ;; perform operation and recur

+            (try

+              (apply (op storage) args)

+              (catch Throwable e

+                (timbre/error "Something unexpected occrus while writing to storage.")

+                (timbre/error e)))

+            (recur (<! writer-chan)))

+          (do

+            (close-queue)

+            (timbre/info "worker queue is cleared. storage worker shuts down")))))

+    ;; return a function that stops the loop

+    (fn []

+      (put! writer-chan {:op :stop})

+      (timbre/info "waiting for storage worker to shut down")

+      (<!! queue-closed?))))

 

 (defn- put-op!

   "Asynchronously put an operation map in a channel."