do-until.clj

gistfile1.txt

(ns do-until-helper.core-test
  (:require [clojure.test :refer :all]
            [do-until-helper.core :refer :all]))


(defn do-async
  [result]
  (future
    (doseq [v [1 2 3 4 5]]
      (swap! result conj v)
      (Thread/sleep 100))))

(defn do-until
  ([func opts]
   (let [stop-time (+ (System/currentTimeMillis)
                      (:timeout opts 5000))]
     (do-until func opts stop-time)))
  ([func opts stop-time]
   (let [mather (:mather opts)
         result (func)]
     (cond
       (> (System/currentTimeMillis) stop-time)
       (throw (ex-info "Do until timeout" {:result result}))

       (mather result) result

       :else (do
               (Thread/sleep (:interval opts 200))
               (do-until func opts stop-time))))))

(deftest async-test
  (let [result (atom [])]

    ;; send message to kafka
    (do-async result)

    (do-until
      (fn []
        ;; do a query to DB to check if records are processed
        @result)
      {:mather (fn [value]
                 ;; looking for a particual type of message
                 (= 5 (count value)))})

    ;; do normal assertions
    (is (= 5 (count @result)))))