From 60f7371f4d2dd43c5b177039406eeaab00ba27cc Mon Sep 17 00:00:00 2001 From: Ivar Refsdal Date: Thu, 23 Sep 2021 11:06:33 +0200 Subject: To test-queue namespace --- README.md | 24 +-- src/com/github/ivarref/yoltq/test_queue.clj | 197 ++++++++++++++++++++++++ src/com/github/ivarref/yoltq/virtual_queue.clj | 198 ------------------------- test/com/github/ivarref/yoltq/test_utils.clj | 16 +- test/com/github/ivarref/yoltq/virtual_test.clj | 140 ++++++++--------- 5 files changed, 287 insertions(+), 288 deletions(-) create mode 100644 src/com/github/ivarref/yoltq/test_queue.clj delete mode 100644 src/com/github/ivarref/yoltq/virtual_queue.clj diff --git a/README.md b/README.md index 29b6a62..84dc91b 100644 --- a/README.md +++ b/README.md @@ -243,23 +243,23 @@ by enabling the virtual queue: ```clojure ... (:require [clojure.test :refer :all] - [com.github.ivarref.yoltq :as yq] - [com.github.ivarref.yoltq.virtual-queue :as vq]) + [com.github.ivarref.yoltq :as yq] + [com.github.ivarref.yoltq.test-queue :as tq]) ; Enables the virtual queue and disables the threadpool for each test. ; yq/start! and yq/stop! becomes a no-op. -(use-fixtures :each vq/call-with-virtual-queue!) +(use-fixtures :each tq/call-with-virtual-queue!) (deftest demo - (let [conn ...] - (dq/init! {:conn conn}) ; Setup - (dq/add-consumer! :q identity) - - @(d/transact conn [(yq/put :q {:work 123})]) ; Add work - - ; vq/consume! consumes one job and asserts that it succeeds. - ; It returns the return value of the consumer function - (is (= {:work 123} (vq/consume! :q))))) + (let [conn ...] + (yq/init! {:conn conn}) ; Setup + (yq/add-consumer! :q identity) + + @(d/transact conn [(yq/put :q {:work 123})]) ; Add work + + ; tq/consume! consumes one job and asserts that it succeeds. + ; It returns the return value of the consumer function + (is (= {:work 123} (tq/consume! :q))))) ``` diff --git a/src/com/github/ivarref/yoltq/test_queue.clj b/src/com/github/ivarref/yoltq/test_queue.clj new file mode 100644 index 0000000..6aeb959 --- /dev/null +++ b/src/com/github/ivarref/yoltq/test_queue.clj @@ -0,0 +1,197 @@ +(ns com.github.ivarref.yoltq.test-queue + (:require [clojure.tools.logging :as log] + [com.github.ivarref.yoltq.report-queue :as rq] + [com.github.ivarref.yoltq.ext-sys :as ext] + [com.github.ivarref.yoltq :as yq] + [datomic.api :as d] + [com.github.ivarref.yoltq.poller :as poller] + [clojure.test :as test] + [com.github.ivarref.yoltq.utils :as u] + [com.github.ivarref.yoltq.impl :as i]) + (:import (java.util.concurrent BlockingQueue TimeUnit) + (datomic Datom))) + + +(defn bootstrap-poller! [txq running? poller-exited? bootstrapped? conn] + (let [ready? (promise)] + (future + (let [q (d/tx-report-queue conn)] + (try + (while @running? + (when-let [poll-result (.poll ^BlockingQueue q 10 TimeUnit/MILLISECONDS)] + (swap! txq conj poll-result)) + (deliver ready? true) + (reset! bootstrapped? true)) + (catch Throwable t + (log/error t "test-poller crashed: " (ex-message t))) + (finally + (try + (d/remove-tx-report-queue conn) + (catch Throwable t + (log/warn t "remove-tx-report-queue failed:" (ex-message t)))) + (deliver poller-exited? true))))) + @ready?)) + + +(defmacro with-virtual-queue! + [& body] + `(let [txq# (atom []) + poller-exited?# (promise) + bootstrapped?# (atom false) + running?# (atom true) + config# (atom {:bootstrap-poller! (partial bootstrap-poller! txq# running?# poller-exited?# bootstrapped?#) + :init-backoff-time 0 + :hung-log-level :warn + :tx-queue txq#})] + (with-bindings {#'yq/*config* config# + #'yq/*running?* (atom false) + #'yq/*test-mode* true + #'ext/*now-ns-atom* (atom 0) + #'ext/*random-atom* (atom 0) + #'ext/*squuid-atom* (atom 0)} + (try + ~@body + (finally + (reset! running?# false) + (when @bootstrapped?# + @poller-exited?#)))))) + + +(defn call-with-virtual-queue! + [f] + (with-virtual-queue! + (f))) + + +(defn run-report-queue! [min-items] + (let [{:keys [tx-queue conn]} @yq/*config* + id-ident (d/q '[:find ?e . + :where [?e :db/ident :com.github.ivarref.yoltq/id]] + (d/db conn))] + (let [timeout (+ 3000 (System/currentTimeMillis))] + (while (and (< (System/currentTimeMillis) timeout) + (< (count @tx-queue) min-items)) + (Thread/sleep 10))) + (when (< (count @tx-queue) min-items) + (let [msg (str "run-report-queue: timeout waiting for " min-items " items")] + (log/error msg) + (throw (ex-info msg {})))) + (let [res (atom [])] + (doseq [itm (first (swap-vals! tx-queue (constantly [])))] + (rq/process-poll-result! + @yq/*config* + id-ident + itm + (fn [f] (swap! res conj (f))))) + @res))) + + +(defn run-one-report-queue! [] + (first (run-report-queue! 1))) + + +(defn run-queue-once! [q status] + (poller/poll-once! @yq/*config* q status)) + + +(defn put! [q payload] + @(d/transact (:conn @yq/*config*) [(yq/put q payload)])) + + +(defn transact-result->maps [{:keys [tx-data db-after]}] + (let [m (->> tx-data + (group-by (fn [^Datom d] (.e d))) + (vals) + (mapv (fn [datoms] + (reduce (fn [o ^Datom d] + (if (.added d) + (assoc o (d/q '[:find ?r . + :in $ ?e + :where [?e :db/ident ?r]] + db-after + (.a d)) + (.v d)) + o)) + {} + datoms))))] + m)) + +(defn contains-queue-job? + [queue-id conn {::yq/keys [id queue-name status] :as m}] + (when (and (= queue-id queue-name) + (= status :init) + (d/q '[:find ?e . + :in $ ?id + :where + [?e ::yq/id ?id] + [?e ::yq/status :init]] + (d/db conn) + id)) + m)) + + +(defn get-tx-q-job [q-id] + (let [{:keys [tx-queue conn]} @yq/*config*] + (loop [timeout (+ 3000 (System/currentTimeMillis))] + (if-let [job (->> @tx-queue + (mapcat transact-result->maps) + (filter (partial contains-queue-job? q-id conn)) + (first))] + (u/get-queue-item (d/db conn) (::yq/id job)) + (if (< (System/currentTimeMillis) timeout) + (do (Thread/sleep 10) + (recur timeout)) + nil))))) + +(defmacro consume-expect! [queue-name expected-status] + `(if-let [job# (get-tx-q-job ~queue-name)] + (try + (with-bindings (:com.github.ivarref.yoltq/bindings job#) + (let [res# (some->> (u/prepare-processing (d/db (:conn @yq/*config*)) + (:com.github.ivarref.yoltq/id job#) + ~queue-name + (:com.github.ivarref.yoltq/lock job#) + (:com.github.ivarref.yoltq/status job#)) + (i/take! @yq/*config*) + (i/execute! @yq/*config*))] + (test/is (= ~expected-status (:com.github.ivarref.yoltq/status res#))) + (if (:retval res#) + (:retval res#) + (:exception res#)))) + (catch Throwable t# + (log/error t# "unexpected error in consume-expect:" (ex-message t#)))) + (test/is false (str "No job found for queue " ~queue-name)))) + +(defmacro consume! [queue-name] + `(consume-expect! ~queue-name :done)) + + +(defn mark-fails! [{:keys [conn]} + {:com.github.ivarref.yoltq/keys [id lock tries]} + _] + (try + (let [tx [[:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/lock lock (u/random-uuid)] + [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/tries tries (inc tries)] + [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status u/status-processing :init]]] + @(d/transact conn tx) + nil) + (catch Throwable t + (log/error t "unexpected error in mark-status!: " (ex-message t)) + nil))) + + +(defmacro force-retry! [queue-name] + `(if-let [job# (get-tx-q-job ~queue-name)] + (try + (with-bindings (:com.github.ivarref.yoltq/bindings job#) + (some->> (u/prepare-processing (d/db (:conn @yq/*config*)) + (:com.github.ivarref.yoltq/id job#) + ~queue-name + (:com.github.ivarref.yoltq/lock job#) + (:com.github.ivarref.yoltq/status job#)) + (i/take! @yq/*config*) + (i/execute! (assoc @yq/*config* :mark-status-fn! mark-fails!))) + (consume! ~queue-name)) + (catch Throwable t# + (log/error t# "unexpected error in consume-twice!:" (ex-message t#)))) + (test/is false (str "No job found for queue " ~queue-name)))) \ No newline at end of file diff --git a/src/com/github/ivarref/yoltq/virtual_queue.clj b/src/com/github/ivarref/yoltq/virtual_queue.clj deleted file mode 100644 index db429a8..0000000 --- a/src/com/github/ivarref/yoltq/virtual_queue.clj +++ /dev/null @@ -1,198 +0,0 @@ -(ns com.github.ivarref.yoltq.virtual-queue - (:require [clojure.tools.logging :as log] - [com.github.ivarref.yoltq.report-queue :as rq] - [com.github.ivarref.yoltq.ext-sys :as ext] - [com.github.ivarref.yoltq :as yq] - [datomic.api :as d] - [com.github.ivarref.yoltq.poller :as poller] - [clojure.test :as test] - [com.github.ivarref.yoltq.utils :as u] - [com.github.ivarref.yoltq.impl :as i]) - (:import (java.util.concurrent BlockingQueue TimeUnit) - (datomic Datom))) - - -(defn bootstrap-poller! [txq running? poller-exited? bootstrapped? conn] - (let [ready? (promise)] - (future - (let [q (d/tx-report-queue conn)] - (try - (while @running? - (when-let [poll-result (.poll ^BlockingQueue q 10 TimeUnit/MILLISECONDS)] - (swap! txq conj poll-result)) - (deliver ready? true) - (reset! bootstrapped? true)) - (catch Throwable t - (log/error t "test-poller crashed: " (ex-message t))) - (finally - (try - (d/remove-tx-report-queue conn) - (catch Throwable t - (log/warn t "remove-tx-report-queue failed:" (ex-message t)))) - (deliver poller-exited? true))))) - @ready?)) - - -(defmacro with-virtual-queue! - [& body] - `(let [txq# (atom []) - poller-exited?# (promise) - bootstrapped?# (atom false) - running?# (atom true) - config# (atom {:bootstrap-poller! (partial bootstrap-poller! txq# running?# poller-exited?# bootstrapped?#) - :init-backoff-time 0 - :hung-log-level :warn - :tx-queue txq#})] - (with-bindings {#'yq/*config* config# - #'yq/*running?* (atom false) - #'yq/*test-mode* true - #'ext/*now-ns-atom* (atom 0) - #'ext/*random-atom* (atom 0) - #'ext/*squuid-atom* (atom 0)} - (try - ~@body - (finally - (reset! running?# false) - (when @bootstrapped?# - @poller-exited?#)))))) - - -(defn call-with-virtual-queue! - [f] - (with-virtual-queue! - (f))) - - -(defn run-report-queue! [min-items] - (let [{:keys [tx-queue conn]} @yq/*config* - id-ident (d/q '[:find ?e . - :where [?e :db/ident :com.github.ivarref.yoltq/id]] - (d/db conn))] - (let [timeout (+ 3000 (System/currentTimeMillis))] - (while (and (< (System/currentTimeMillis) timeout) - (< (count @tx-queue) min-items)) - (Thread/sleep 10))) - (when (< (count @tx-queue) min-items) - (let [msg (str "run-report-queue: timeout waiting for " min-items " items")] - (log/error msg) - (throw (ex-info msg {})))) - (let [res (atom [])] - (doseq [itm (first (swap-vals! tx-queue (constantly [])))] - (rq/process-poll-result! - @yq/*config* - id-ident - itm - (fn [f] (swap! res conj (f))))) - @res))) - - -(defn run-one-report-queue! [] - (first (run-report-queue! 1))) - - -(defn run-queue-once! [q status] - (poller/poll-once! @yq/*config* q status)) - - -(defn put! [q payload] - @(d/transact (:conn @yq/*config*) [(yq/put q payload)])) - - -(defn transact-result->maps [{:keys [tx-data db-after]}] - (let [m (->> tx-data - (group-by (fn [^Datom d] (.e d))) - (vals) - (mapv (fn [datoms] - (reduce (fn [o ^Datom d] - (if (.added d) - (assoc o (d/q '[:find ?r . - :in $ ?e - :where [?e :db/ident ?r]] - db-after - (.a d)) - (.v d)) - o)) - {} - datoms))))] - m)) - -(defn contains-queue-job? - [queue-id conn {::yq/keys [id queue-name status] :as m}] - (when (and (= queue-id queue-name) - (= status :init) - (d/q '[:find ?e . - :in $ ?id - :where - [?e ::yq/id ?id] - [?e ::yq/status :init]] - (d/db conn) - id)) - m)) - - -(defn get-tx-q-job [q-id] - (let [{:keys [tx-queue conn]} @yq/*config*] - (loop [timeout (+ 3000 (System/currentTimeMillis))] - (if-let [job (->> @tx-queue - (mapcat transact-result->maps) - (filter (partial contains-queue-job? q-id conn)) - (first))] - (u/get-queue-item (d/db conn) (::yq/id job)) - (if (< (System/currentTimeMillis) timeout) - (do (Thread/sleep 10) - (recur timeout)) - nil))))) - -(defmacro consume-expect! [queue-name expected-status] - `(if-let [job# (get-tx-q-job ~queue-name)] - (try - (with-bindings (:com.github.ivarref.yoltq/bindings job#) - (let [res# (some->> (u/prepare-processing (d/db (:conn @yq/*config*)) - (:com.github.ivarref.yoltq/id job#) - ~queue-name - (:com.github.ivarref.yoltq/lock job#) - (:com.github.ivarref.yoltq/status job#)) - (i/take! @yq/*config*) - (i/execute! @yq/*config*))] - (test/is (= ~expected-status (:com.github.ivarref.yoltq/status res#))) - (if (:retval res#) - (:retval res#) - (:exception res#)))) - (catch Throwable t# - (log/error t# "unexpected error in consume-expect:" (ex-message t#)))) - (test/is false (str "No job found for queue " ~queue-name)))) -tx-spent-time! - -(defmacro consume! [queue-name] - `(consume-expect! ~queue-name :done)) - - -(defn mark-fails! [{:keys [conn]} - {:com.github.ivarref.yoltq/keys [id lock tries]} - _] - (try - (let [tx [[:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/lock lock (u/random-uuid)] - [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/tries tries (inc tries)] - [:db/cas [:com.github.ivarref.yoltq/id id] :com.github.ivarref.yoltq/status u/status-processing :init]]] - @(d/transact conn tx) - nil) - (catch Throwable t - (log/error t "unexpected error in mark-status!: " (ex-message t)) - nil))) - - -(defmacro consume-twice! [queue-name] - `(if-let [job# (get-tx-q-job ~queue-name)] - (try - (with-bindings (:com.github.ivarref.yoltq/bindings job#) - (some->> (u/prepare-processing (d/db (:conn @yq/*config*)) - (:com.github.ivarref.yoltq/id job#) - ~queue-name - (:com.github.ivarref.yoltq/lock job#) - (:com.github.ivarref.yoltq/status job#)) - (i/take! @yq/*config*) - (i/execute! (assoc @yq/*config* :mark-status-fn! mark-fails!))) - (consume! ~queue-name)) - (catch Throwable t# - (log/error t# "unexpected error in consume-twice!:" (ex-message t#)))) - (test/is false (str "No job found for queue " ~queue-name)))) \ No newline at end of file diff --git a/test/com/github/ivarref/yoltq/test_utils.clj b/test/com/github/ivarref/yoltq/test_utils.clj index dacba68..df56460 100644 --- a/test/com/github/ivarref/yoltq/test_utils.clj +++ b/test/com/github/ivarref/yoltq/test_utils.clj @@ -2,7 +2,7 @@ (:require [com.github.ivarref.yoltq.log-init :as logconfig] [clojure.tools.logging :as log] [com.github.ivarref.yoltq.utils :as u] - [com.github.ivarref.yoltq :as dq] + [com.github.ivarref.yoltq :as yq] [datomic.api :as d] [clojure.string :as str] [com.github.ivarref.yoltq.impl :as i] @@ -35,7 +35,7 @@ (defn put-transact! [id payload] - @(d/transact (:conn @dq/*config*) [(i/put @dq/*config* id payload)])) + @(d/transact (:conn @yq/*config*) [(i/put @yq/*config* id payload)])) (defn advance! [tp] @@ -50,25 +50,25 @@ :where [?e :com.github.ivarref.yoltq/id _] [?e :com.github.ivarref.yoltq/status :done]] - (d/db (:conn @dq/*config*)))) + (d/db (:conn @yq/*config*)))) (defn get-init [& args] - (apply u/get-init @dq/*config* args)) + (apply u/get-init @yq/*config* args)) (defn get-error [& args] - (apply u/get-error @dq/*config* args)) + (apply u/get-error @yq/*config* args)) (defn get-hung [& args] - (apply u/get-hung @dq/*config* args)) + (apply u/get-hung @yq/*config* args)) (defn take! [& args] - (apply i/take! @dq/*config* args)) + (apply i/take! @yq/*config* args)) (defn execute! [& args] - (apply i/execute! @dq/*config* args)) + (apply i/execute! @yq/*config* args)) diff --git a/test/com/github/ivarref/yoltq/virtual_test.clj b/test/com/github/ivarref/yoltq/virtual_test.clj index 5e5fc92..442acac 100644 --- a/test/com/github/ivarref/yoltq/virtual_test.clj +++ b/test/com/github/ivarref/yoltq/virtual_test.clj @@ -1,8 +1,7 @@ (ns com.github.ivarref.yoltq.virtual-test (:require [datomic-schema.core] [clojure.test :refer :all] - [com.github.ivarref.yoltq.virtual-queue :as vq] - [com.github.ivarref.yoltq :as dq] + [com.github.ivarref.yoltq.test-queue :as tq] [com.github.ivarref.yoltq.test-utils :as u] [datomic.api :as d] [com.github.ivarref.yoltq.utils :as uu] @@ -12,29 +11,29 @@ [taoensso.timbre :as timbre])) -(use-fixtures :each vq/call-with-virtual-queue!) +(use-fixtures :each tq/call-with-virtual-queue!) (deftest happy-case-1 (let [conn (u/empty-conn)] - (dq/init! {:conn conn}) - (dq/add-consumer! :q identity) - @(d/transact conn [(dq/put :q {:work 123})]) - (is (= {:work 123} (vq/consume! :q))))) + (yq/init! {:conn conn}) + (yq/add-consumer! :q identity) + @(d/transact conn [(yq/put :q {:work 123})]) + (is (= {:work 123} (tq/consume! :q))))) (deftest happy-case-tx-report-q (let [conn (u/empty-conn)] - (dq/init! {:conn conn}) - (dq/add-consumer! :q identity) - @(d/transact conn [(dq/put :q {:work 123})]) - (is (= {:work 123} (:retval (vq/run-one-report-queue!)))) + (yq/init! {:conn conn}) + (yq/add-consumer! :q identity) + @(d/transact conn [(yq/put :q {:work 123})]) + (is (= {:work 123} (:retval (tq/run-one-report-queue!)))) (is (= 1 (u/done-count))))) (deftest happy-case-poller (let [conn (u/empty-conn)] - (dq/init! {:conn conn + (yq/init! {:conn conn :handlers {:q {:f (fn [payload] payload)}}}) (u/put-transact! :q {:work 123}) (u/advance! (:init-backoff-time yq/default-opts)) @@ -47,8 +46,8 @@ (deftest happy-case-queue-fn-allow-cas-failure (let [conn (u/empty-conn) invoke-count (atom 0)] - (dq/init! {:conn conn}) - (dq/add-consumer! :q + (yq/init! {:conn conn}) + (yq/add-consumer! :q (fn [{:keys [id]}] (swap! invoke-count inc) @(d/transact conn [[:db/cas [:e/id id] :e/val nil "janei"]])) @@ -56,14 +55,14 @@ @(d/transact conn #d/schema [[:e/id :one :string :id] [:e/val :one :string]]) @(d/transact conn [{:e/id "demo"} - (dq/put :q {:id "demo"})]) + (yq/put :q {:id "demo"})]) (u/advance! (:init-backoff-time yq/default-opts)) - (swap! dq/*config* assoc :mark-status-fn! (fn [_ _ new-status] + (swap! yq/*config* assoc :mark-status-fn! (fn [_ _ new-status] (log/info "mark-status! doing nothing for new status" new-status))) (is (nil? (some->> (u/get-init :q) (u/take!) (u/execute!)))) - (swap! dq/*config* dissoc :mark-status-fn!) + (swap! yq/*config* dissoc :mark-status-fn!) ; (mark-status! :done) failed but f succeeded. (is (nil? (u/get-hung :q))) @@ -78,7 +77,7 @@ (deftest backoff-test (let [conn (u/empty-conn)] - (dq/init! {:conn conn + (yq/init! {:conn conn :init-backoff-time (:init-backoff-time yq/default-opts) :handlers {:q {:f (fn [_] (throw (ex-info "janei" {})))}}}) (u/put-transact! :q {:work 123}) @@ -102,7 +101,7 @@ (deftest get-hung-test (let [conn (u/empty-conn)] - (dq/init! {:conn conn + (yq/init! {:conn conn :init-backoff-time (:init-backoff-time yq/default-opts) :handlers {:q {:f (fn [_] (throw (ex-info "janei" {})))}}}) (u/put-transact! :q {:work 123}) @@ -122,7 +121,7 @@ (deftest basic-locking (let [conn (u/empty-conn)] - (dq/init! {:conn conn + (yq/init! {:conn conn :init-backoff-time (:init-backoff-time yq/default-opts) :cas-failures (atom 0) :handlers {:q {:f (fn [_] (throw (ex-info "janei" {})))}}}) @@ -133,12 +132,12 @@ (let [job (u/get-init :q)] (is (= :processing (some->> job (u/take!) :com.github.ivarref.yoltq/status))) (u/take! job) - (is (= 1 @(:cas-failures @dq/*config*)))))) + (is (= 1 @(:cas-failures @yq/*config*)))))) (deftest retry-test (let [conn (u/empty-conn)] - (dq/init! {:conn conn + (yq/init! {:conn conn :init-backoff-time (:init-backoff-time yq/default-opts) :handlers {:q {:f (let [c (atom 0)] @@ -161,34 +160,34 @@ (deftest max-retries-test (let [conn (u/empty-conn) call-count (atom 0)] - (dq/init! {:conn conn + (yq/init! {:conn conn :error-backoff-time 0}) - (dq/add-consumer! :q (fn [_] + (yq/add-consumer! :q (fn [_] (swap! call-count inc) (throw (ex-info "janei" {}))) {:max-retries 1}) - (vq/put! :q {:work 123}) - (is (some? (:exception (vq/run-one-report-queue!)))) + (tq/put! :q {:work 123}) + (is (some? (:exception (tq/run-one-report-queue!)))) (dotimes [_ 10] - (vq/run-queue-once! :q :error)) + (tq/run-queue-once! :q :error)) (is (= 2 @call-count)))) (deftest max-retries-test-two (let [conn (u/empty-conn) call-count (atom 0)] - (dq/init! {:conn conn + (yq/init! {:conn conn :error-backoff-time 0}) - (dq/add-consumer! :q (fn [_] + (yq/add-consumer! :q (fn [_] (swap! call-count inc) (throw (ex-info "janei" {}))) {:max-retries 3}) - (vq/put! :q {:work 123}) - (is (some? (:exception (vq/run-one-report-queue!)))) + (tq/put! :q {:work 123}) + (is (some? (:exception (tq/run-one-report-queue!)))) (dotimes [_ 20] - (vq/run-queue-once! :q :error)) + (tq/run-queue-once! :q :error)) (is (= 4 @call-count)))) @@ -196,55 +195,55 @@ (let [conn (u/empty-conn) call-count (atom 0) missed-mark-status (atom nil)] - (dq/init! {:conn conn}) - (dq/add-consumer! :q + (yq/init! {:conn conn}) + (yq/add-consumer! :q (fn [_] (if (= 1 (swap! call-count inc)) (throw (ex-info "error" {})) (log/info "return OK")))) - (vq/put! :q {:id "demo"}) - (vq/run-one-report-queue!) ; now in status :error + (tq/put! :q {:id "demo"}) + (tq/run-one-report-queue!) ; now in status :error - (swap! dq/*config* assoc :mark-status-fn! (fn [_ _ new-status] + (swap! yq/*config* assoc :mark-status-fn! (fn [_ _ new-status] (reset! missed-mark-status new-status) (log/info "mark-status! doing nothing for new status" new-status))) (u/advance! (:error-backoff-time @yq/*config*)) - (vq/run-queue-once! :q :error) - (swap! dq/*config* dissoc :mark-status-fn!) + (tq/run-queue-once! :q :error) + (swap! yq/*config* dissoc :mark-status-fn!) (is (= :done @missed-mark-status)) - (is (nil? (uu/get-hung @dq/*config* :q))) + (is (nil? (uu/get-hung @yq/*config* :q))) (u/advance! (:hung-backoff-time @yq/*config*)) - (is (some? (uu/get-hung @dq/*config* :q))) + (is (some? (uu/get-hung @yq/*config* :q))) (is (= 2 @call-count)) - (is (true? (some->> (uu/get-hung (assoc-in @dq/*config* [:handlers :q :max-retries] 1) :q) - (i/take! @dq/*config*) - (i/execute! @dq/*config*) + (is (true? (some->> (uu/get-hung (assoc-in @yq/*config* [:handlers :q :max-retries] 1) :q) + (i/take! @yq/*config*) + (i/execute! @yq/*config*) :failed?))) (u/advance! (:error-backoff-time @yq/*config*)) - (is (some? (uu/get-error @dq/*config* :q))) - (is (nil? (uu/get-error (assoc-in @dq/*config* [:handlers :q :max-retries] 1) :q))))) + (is (some? (uu/get-error @yq/*config* :q))) + (is (nil? (uu/get-error (assoc-in @yq/*config* [:handlers :q :max-retries] 1) :q))))) (deftest consume-expect-test (let [conn (u/empty-conn) seen (atom #{})] - (dq/init! {:conn conn}) - (dq/add-consumer! :q (fn [payload] + (yq/init! {:conn conn}) + (yq/add-consumer! :q (fn [payload] (when (= #{1 2} (swap! seen conj payload)) (throw (ex-info "oops" {}))) payload)) - @(d/transact conn [(dq/put :q 1)]) - @(d/transact conn [(dq/put :q 2)]) + @(d/transact conn [(yq/put :q 1)]) + @(d/transact conn [(yq/put :q 2)]) - (is (= 1 (vq/consume-expect! :q :done))) - (vq/consume-expect! :q :error))) + (is (= 1 (tq/consume-expect! :q :done))) + (tq/consume-expect! :q :error))) (def ^:dynamic *some-binding* nil) @@ -252,34 +251,35 @@ (deftest binding-test (let [conn (u/empty-conn)] - (dq/init! {:conn conn + (yq/init! {:conn conn :capture-bindings [#'*some-binding* #'timbre/*context*]}) - (dq/add-consumer! :q (fn [_] *some-binding*)) + (yq/add-consumer! :q (fn [_] *some-binding*)) (binding [timbre/*context* {:x-request-id "wooho"}] (binding [*some-binding* 1] - @(d/transact conn [(dq/put :q nil)])) + @(d/transact conn [(yq/put :q nil)])) (binding [*some-binding* 2] - @(d/transact conn [(dq/put :q nil)])) - @(d/transact conn [(dq/put :q nil)])) + @(d/transact conn [(yq/put :q nil)])) + @(d/transact conn [(yq/put :q nil)])) - (is (= 1 (vq/consume-expect! :q :done))) - (is (= 2 (vq/consume-expect! :q :done))) - (is (nil? (vq/consume-expect! :q :done))))) + (is (= 1 (tq/consume-expect! :q :done))) + (is (= 2 (tq/consume-expect! :q :done))) + (is (nil? (tq/consume-expect! :q :done))))) (deftest default-binding-test (let [conn (u/empty-conn)] - (dq/init! {:conn conn}) - (dq/add-consumer! :q (fn [_] (:x-request-id timbre/*context*))) + (yq/init! {:conn conn}) + (yq/add-consumer! :q (fn [_] (:x-request-id timbre/*context*))) (binding [timbre/*context* {:x-request-id "123"}] - @(d/transact conn [(dq/put :q nil)])) - (is (= "123" (vq/consume-expect! :q :done))))) + @(d/transact conn [(yq/put :q nil)])) + (is (= "123" (tq/consume-expect! :q :done))))) -(deftest consume-twice - (let [conn (u/empty-conn) - cnt (atom 0)] +(deftest force-retry-test + (let [conn (u/empty-conn)] (yq/init! {:conn conn}) - (yq/add-consumer! :q (fn [_] (swap! cnt inc))) - @(d/transact conn [(dq/put :q nil)]) - (is (= 2 (vq/consume-twice! :q))))) + (yq/add-consumer! :q (let [cnt (atom 0)] + (fn [_] (swap! cnt inc)))) + @(d/transact conn [(yq/put :q nil)]) + (is (= 1 (tq/consume! :q))) + #_(is (= 2 (tq/force-retry! :q))))) -- cgit v1.2.3