notid/integreat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alternative requset listener

Browse Source
This commit is contained in:
Bryce Covert
2021-12-18 19:28:58 -08:00
parent 64720a4bd6
commit 88f070550e
1 changed files with 15 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
src/clj/auto_ap/background/requests.clj
View File

@@ -7,36 +7,25 @@
[auto-ap.intuit.import :as i]
[unilog.context :as lc]))
(def stopped? (atom false))
(defn process-1 []
(let [[{:keys [message-id receipt-handle body]}] (:messages (sqs/receive-message {:queue-url (:requests-queue-url env)
:count 1}))]
(when message-id
(log/infof "processing message %s with body %s" message-id body )
(if (= ":intuit" body)
(try
(i/upsert-transactions)
(catch Exception e
(log/error e))))
(sqs/delete-message {:queue-url (:requests-queue-url env)
:receipt-handle receipt-handle} ))))
(lc/with-context {:source "Request poller"}
(log/info "Checking SQS...")
(let [[{:keys [message-id receipt-handle body]}] (:messages (sqs/receive-message {:queue-url (:requests-queue-url env)
:count 1}))]
(when message-id
(log/infof "processing message %s with body %s" message-id body )
(if (= ":intuit" body)
(try
(i/upsert-transactions)
(catch Exception e
(log/error e))))
(sqs/delete-message {:queue-url (:requests-queue-url env)
:receipt-handle receipt-handle} )))))
(defn fake-message []
(sqs/send-message {:queue-url (:requests-queue-url env)
:message-body ":intuit"} ))
(defn listen-sqs []
(lc/with-context {:source "Request poller"}
(loop []
(process-1)
(log/info "Checking SQS....")
(Thread/sleep 5000)
(when-not @stopped?
(log/info "stopping")
(recur)))))
(mount/defstate request-listener
:start (future (listen-sqs))
:stop (reset! stopped? true))
:start (scheduler/every (* 1000 5) process-1)
:stop (scheduler/stop request-listener))