redisqueue.clj

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

(def server1-conn {:pool {} :spec {:host "localhost" :port 6379}}) ;; XXX config here                                                                                                                                                         

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

(defn create-worker
  [queue handler-fn & {:keys [server-spec monitor-fn nthreads]
                       :or {server-spec server1-conn
                            monitor-fn nil
                            nthreads  (.availableProcessors (Runtime/getRuntime))}}]
  (car-mq/worker server-spec
                 queue
                 {:handler handler-fn
                  :throttle-ms 150
                  :nthreads nthreads
                  :eoq-backoff-ms 500
                  :monitor monitor-fn}))

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

(defn enqueue
  [queue message]
  (wcar* (car-mq/enqueue queue message)))