core, processor: refactor startup; graceful shutdown

- implement a functional rewrite of core.clj, no more `def`, no more
states. callable from an REPL
- handle SIGINT/SIGTERM and perform graceful shutdown
- print clear logs when starting up and shutting down

3 files changed, 108 insertions, 58 deletions

diff --git a/src/weepinbell/core.clj b/src/weepinbell/core.clj
index 4c867ef..034bb96 100644
--- a/src/weepinbell/core.clj
+++ b/src/weepinbell/core.clj
@@ -7,38 +7,69 @@
             [weepinbell.config :refer [load-config]]
             [weepinbell.webhook :as webhook]
             [weepinbell.storage.edn-storage :as edn]
-            [clojure.core.async :refer [chan]]
+            [clojure.core.async :refer [chan >!! <!!]]
             [taoensso.timbre :as timbre]
             [weepinbell.storage.spec])
+  (:import (sun.misc Signal SignalHandler))
   (:gen-class)) ;; Essential for "lein uberjar"
 
-(def ^{:doc "Webmention channel.
-       Receiver puts webmention requests in. Processor takes them out."}
-  wmc (chan))
+; Handle startup; Callable from REPL
+;; processor starter is in ns weepinbell.proceesor.starter
+;; webserver starter is defined here, temporarily
+(defn start-webserver
+  "Starts web server. Returns a function that stops the server."
+  [server-meta]
+  (let [{:keys [wmc store config]} server-meta
+        port (Integer/parseInt (or (:port config) (System/getenv "PORT") "3000"))]
+    (timbre/info "starting web server, listening on" port)
+    (run-server (app wmc store config) {:port port})))
 
-(def ^{:doc "Weepinbell configuration, 
-       loaded from config.edn and merged with defaults."}
-  config (load-config))
-
-(def ^{:doc "Storage middle layer, a map of data-manipulating functions.
-       Supports various storage as long as the map conforms storage.spec."}
-  store (s/conform :weepinbell.storage.spec/vtable
-                   (edn/use-store (:data-path config))))
-
-(def ^{:doc "Hooks, a map of callback functions that processor calls 
-       on certain events. Can be used for webhooks and more."}
-  hooks (webhook/make-hooks config))
+;; start whole system
+(defn start-system
+  "Start webmention processor and webserver in the background.
+   Returns a function that performs a graceful shutdown."
+  []
+  (let [;; webmention channel 
+        ;; Receiver puts webmention requests in. 
+        ;; Processor takes them out.
+        wmc (chan)
+        ;; Weepinbell configuration, 
+        ;; loaded from config.edn and merged with defaults.
+        config (load-config)
+        ;; Storage middle layer, a map of data-manipulating functions.
+        ;; Supports various storage as long as the map conforms storage.spec.
+        store (s/conform :weepinbell.storage.spec/vtable
+                         (edn/use-store (:data-path config)))
+        ;; Hooks, a map of callback functions that processor calls 
+        ;; on certain events. Can be used for webhooks and more.
+        hooks (webhook/make-hooks config)
+        ;; start processor and webservers and get stoppers 
+        stop-processor (start-processor
+                        {:wmc wmc
+                         :callbacks hooks
+                         :storage store
+                         :config config})
+        stop-webserver (start-webserver
+                        {:wmc wmc
+                         :store store
+                         :config config})]
+    (fn [] (stop-processor) (stop-webserver))))
 
+; Entry point; Terminal user interface
+;; start up system and wait for signals to perform graceful shutdown
 (defn -main [& _]
-  ;; Start webmention processor
-  (start-processor 
-    {:wmc wmc 
-     :callbacks hooks 
-     :storage store 
-     :config config})
-  ;; Start receiver server
-  ;; Check for a PORT environment variable, default to 3000
-  (let [port (Integer/parseInt (or (:port config) (System/getenv "PORT") "3000"))]
-    (run-server (app wmc store config) {:port port :join? true})
-    (timbre/info "weepinbell started, listening on" port)))
+  (timbre/info "starting weepinbell...")
+  (let [done-chan (chan 1)
+        stop-system (start-system)]
+    ;; Register signal handlers for SIGINT and SIGTERM
+    (doseq [signal ["INT" "TERM"]]
+      (Signal/handle (Signal. signal) 
+        (reify SignalHandler
+          (handle [_ sig] 
+            (stop-system)
+            (timbre/info "signal" (.getName sig) "received." 
+                         "gracefully shut down weepinbell.")
+            (>!! done-chan true)))))
+    ;; block til signal received
+    (<!! done-chan)))
 
diff --git a/src/weepinbell/processor/starter.clj b/src/weepinbell/processor/starter.clj
index 384b4ac..ebf2f17 100644
--- a/src/weepinbell/processor/starter.clj
+++ b/src/weepinbell/processor/starter.clj
@@ -1,6 +1,6 @@
 (ns weepinbell.processor.starter

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

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

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

             [taoensso.timbre :as timbre]

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

             [weepinbell.storage.spec]

@@ -23,17 +23,24 @@
   [metadata]

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

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

+    ;; start processing loop

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

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

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

             id (digest (str s t))]

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

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

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

+        (if stop? 

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

+          ;; process found webmention

+          (do

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

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

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

+    ;; return a function that stops the loop 

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

diff --git a/test/weepinbell/processor_starter_test.clj b/test/weepinbell/processor_starter_test.clj
index 65f6427..df90995 100644
--- a/test/weepinbell/processor_starter_test.clj
+++ b/test/weepinbell/processor_starter_test.clj
@@ -3,23 +3,35 @@
             [clojure.core.async :refer [chan put!]]

             [weepinbell.processor.starter :refer [start-processor]]))

 

-(def test-wmc (chan))

-(def received (promise))

-(def processor-meta {:wmc test-wmc

-                     :callbacks {:on-receive (fn [_] (deliver received true))

-                                 :on-create (fn [_])

-                                 :on-delete (fn [_])

-                                 :on-update (fn [_])}

-                     :storage {:get (fn [& _])

-                               :delete (fn [& _])

-                               :append (fn [& _])

-                               :update (fn [& _])

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

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

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

+(defn- test-processor [test-wmc received?]

+  {:pre [(instance? clojure.lang.IPending received?)]}

+  (let [pmeta {:wmc test-wmc

+               :callbacks {:on-receive (fn [_] (deliver received? true))

+                           :on-create (fn [_])

+                           :on-delete (fn [_])

+                           :on-update (fn [_])}

+               :storage {:get (fn [& _])

+                         :delete (fn [& _])

+                         :append (fn [& _])

+                         :update (fn [& _])

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

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

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

+    (start-processor pmeta)))

 

 (deftest start-processor-test

-  (testing "starting processor and putting a fake webmention in queue"

-    (start-processor processor-meta)

-    (put! test-wmc {:s "s" :t "t"})

-    (is (deref received 50 false))))

+  (testing "graceful shutdown"

+    (let [test-wmc (chan)

+          received? (promise)

+          shutdown (test-processor test-wmc received?)]

+      (shutdown)

+      (put! test-wmc {:s "s" :t "t"})

+      (is (not (deref received? 50 false)))))

+  (testing "processor starting and taking webmention"

+    (let [test-wmc (chan)

+          received? (promise)

+          shutdown (test-processor test-wmc received?)]

+      (put! test-wmc {:s "s" :t "t"})

+      (is (deref received? 50 false))

+      (shutdown))))

+