couchbase-reader.clj

(defn- db-connect []
  (couchbase/create-client {:bucket "default"
                            :uris ["http://couchbase001.local:8091/pools"
                                   "http://couchbase002.local:8091/pools"
                                   "http://couchbase003.local:8091/pools"
                                   "http://couchbase004.local:8091/pools"]}))

(defn make-key [app-id device-id]
  (str app-id ":" device-id))

(defn start-handler [reqs-channel]
  (dotimes [n 8]
    (future
      (while true
        (try
          (let [fut (async/<!! reqs-channel)]
            (if @fut
              (swap! stats/success-count inc)
              (swap! stats/failure-count inc)))
          (catch Exception ex
            (.printStackTrace ex)))))))

(defn start-client [input-channel]
  (let [db-client (db-connect)
        reqs-channel (async/chan 5000)]
    (start-handler reqs-channel)
    (dotimes [n 8]
      (future
        (while true
          (try
            (let [msg (async/<!! input-channel)
                  app-id (:app_id msg)
                  device-id (:appsflyer_id (:device msg))
                  key (make-key app-id device-id)]
              (async/>!! reqs-channel (couchbase/async-get db-client key)))
            (catch Exception ex
              (.printStackTrace ex))))))))