blob: 1ce367582ce0e5b54eb1c32137e630714528c111 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
(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]
(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."
[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]))
|
