-
Notifications
You must be signed in to change notification settings - Fork 3
/
actions.clj
25 lines (23 loc) · 863 Bytes
/
actions.clj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
(ns piped.actions
(:require [clojure.tools.logging :as log]
[clojure.core.async :as async]
[piped.utils :as utils]
[piped.sqs :as sqs]))
(defn spawn-acker
"Acks batches of messages."
[client input-chan]
(async/go-loop []
(when-some [batch (async/<! input-chan)]
(let [response (async/<! (sqs/ack-many client batch))]
(when (utils/anomaly? response)
(log/error "Error when trying to ack batch of messages." (pr-str response))))
(recur))))
(defn spawn-nacker
"Nacks batches of messages."
[client input-chan]
(async/go-loop []
(when-some [batch (async/<! input-chan)]
(let [response (async/<! (sqs/nack-many client batch))]
(when (utils/anomaly? response)
(log/error "Error when trying to nack batch of messages." (pr-str response))))
(recur))))