core, storage: add storage worker that handles writes asynchronously and thread-safely

fix

2 files changed, 55 insertions, 4 deletions

diff --git a/src/weepinbell/core.clj b/src/weepinbell/core.clj
index 20c0aef..1cc233e 100644
--- a/src/weepinbell/core.clj
+++ b/src/weepinbell/core.clj
@@ -4,6 +4,7 @@
             [weepinbell.config :refer [load-config]]
             [weepinbell.webhook :as webhook]
             [weepinbell.storage.edn-storage :as edn]
+            [weepinbell.storage.worker :as worker]
             [clojure.core.async :refer [chan >!! <!!]]
             [taoensso.timbre :as timbre]
             ;; two main subsystems' facades
@@ -28,7 +29,8 @@
         config (load-config (or (System/getenv "WEEPINBELL_CONFIG") "config.edn"))
         ;; Storage middle layer, a map of data-manipulating functions.
         ;; Supports various storage as long as the map conforms storage.spec.
-        store (edn/use-store (:data-path config))
+        bare-store (edn/use-store (:data-path config))
+        [stop-worker async-store] (worker/start bare-store)
         ;; Hooks, a map of callback functions that processor calls 
         ;; on certain events. Can be used for webhooks and more.
         hooks (webhook/make-hooks config)
@@ -36,13 +38,17 @@
         stop-processor (start-processor
                         {:wmc wmc
                          :callbacks hooks
-                         :storage store
+                         :storage async-store
                          :config config})
         stop-webserver (start-webserver
                         {:wmc wmc
-                         :store store
+                         :store async-store
                          :config config})]
-    (fn [] (stop-processor) (stop-webserver) true)))
+    (fn [] 
+      (stop-processor) 
+      (stop-worker)
+      (stop-webserver) 
+      true)))
 
 ; Entry point; Terminal user interface
 ;; start up system and wait for signals to perform graceful shutdown
diff --git a/src/weepinbell/storage/worker.clj b/src/weepinbell/storage/worker.clj
new file mode 100644
index 0000000..41d4912
--- /dev/null
+++ b/src/weepinbell/storage/worker.clj
@@ -0,0 +1,45 @@
+(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!]]

+            [taoensso.timbre :as timbre]))

+

+(defn- start-worker

+  "Asynchrnously start worker loop that takes from writer-chan 

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

+

+(defn- put-op!

+  "Asynchronously put an operation map in a channel."

+  [channel op args]

+  (put! channel {:op op :args args}))

+

+(defn start

+  "Start worker. Wraps a base storage and returns a vector of 

+  a stopper function and a asynchronous storage vtable."

+  [storage]

+  (let [writer-chan (chan)

+        put-op (partial put-op! writer-chan)

+        stop-worker (start-worker storage writer-chan)

+        vtable {:append #(put-op :append [%])

+                :update (fn [& args] (put-op :update args))

+                :delete (fn [& args] (put-op :delete args))

+                :get (:get storage)

+                :get-by-domain (:get-by-domain storage)}]

+    [stop-worker vtable]))