redis-carmine message queue

(ns project.queue
  (:require [project.error :as e]
            [project.env :refer [env]]

            [taoensso.carmine :as car :refer (wcar)]
            [taoensso.carmine.message-queue :as car-mq]

            [clojure.tools.logging :as log])

  (:refer-clojure :exclude [send]))


(defmulti action
  (fn [data]
    (when (map? data)
      (:action data))))


(defmethod action :default
  [data]
  (log/errorf "MQ warn: unknown data, %s" data))


(def conn {:pool {}
           :spec {:host (:redis-host env)
                  :port (:redis-port env)
                  :user (:redis-user env)
                  :password (:redis-password env)}})


(def queue "queue")


(defn worker-handler
  [{:keys [message mid attempt]}]

  (try
    (action message)
    (log/infof "MQ: mid %s processed, attempt: %s"
               mid attempt)
    {:status :success}

    (catch Throwable e
      (log/errorf "MQ error: %s, mid: %s, message: %s, attempt: %s"
                  (e/exc-msg e) mid message attempt)

      {:status :error :throwable e})))


(defonce worker
  (car-mq/worker
   conn queue
   {:auto-start false
    :monitor
    (fn [{:keys [mid-circle-size ndry-runs poll-reply] :as args}]
      (log/infof "Monitor: %s" args))
    :handler worker-handler}))


(defmacro wcar*
  [& body]
  `(car/wcar conn ~@body))


(defn send
  [data]
  (log/debugf "Message sent: %s" data)
  (wcar* (car-mq/enqueue queue data)))


(defn start
  []
  (car-mq/start worker))

(defn stop
  []
  (car-mq/stop worker))

;;
;; Init
;;

(defn init
  []
  (start))