blob: 1f4e65db79fc7dfa9dbe5745fc3f02d7f36a7f04 (
plain) (
blame)
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
53
54
55
|
(ns com.github.ivarref.yoltq.poller
(:require [com.github.ivarref.yoltq.utils :as u]
[com.github.ivarref.yoltq.impl :as i]
[clojure.tools.logging :as log]))
(defn poll-once! [cfg q status]
(when-let [item (case status
:init (u/get-init cfg q)
:error (u/get-error cfg q)
:hung (u/get-hung cfg q))]
(with-bindings (get item :bindings {})
(some->> item
(i/take! cfg)
(i/execute! cfg)))))
(defn poll-queue! [running?
{:keys [running-queues] :as cfg}
[queue-name status :as q]]
(try
(let [[old _] (swap-vals! running-queues conj q)]
(if-not (contains? old q)
(try
(log/debug "polling queue" queue-name "for status" status)
(let [start-time (u/now-ns)
last-res (loop [prev-res nil]
(when @running?
(let [res (poll-once! cfg queue-name status)]
(if (and res (:success? res))
(recur res)
prev-res))))]
(let [spent-ns (- (u/now-ns) start-time)]
(log/trace "done polling queue" q "in"
(format "%.1f" (double (/ spent-ns 1e6)))
"ms"))
last-res)
(finally
(swap! running-queues disj q)))
(log/debug "queue" q "is already being polled, doing nothing...")))
(catch Throwable t
(log/error t "poll-queue! crashed:" (ex-message t)))
(finally)))
(defn poll-all-queues! [running? config-atom pool]
(try
(when @running?
(let [{:keys [handlers]} @config-atom]
(doseq [q (shuffle (vec (for [q-name (keys handlers)
status [:init :error :hung]]
[q-name status])))]
(.execute pool (fn [] (poll-queue! running? @config-atom q))))))
(catch Throwable t
(log/error t "poll-all-queues! crashed:" (ex-message t)))))
|
