minimal async sqs

core.clj

(ns bandarlog.core
 (:require [clojure.core.async :refer [thread-call <!! thread go <!]])
 (:import com.amazonaws.auth.BasicAWSCredentials
          com.amazonaws.services.sqs.AmazonSQSAsyncClient
          com.amazonaws.services.sqs.buffered.AmazonSQSBufferedAsyncClient
          (com.amazonaws.services.sqs.model SendMessageRequest
                                            SendMessageBatchRequest
                                            SendMessageBatchRequestEntry
                                            ReceiveMessageRequest
                                            ReceiveMessageResult)))

(defn create-client
  "Creates a new Buffered Async Client instance with given credentials."
  [client-id client-secret]
  (->  (BasicAWSCredentials. client-id client-secret)
       (AmazonSQSAsyncClient.)
       (AmazonSQSBufferedAsyncClient.)))

(defn send-message [client queue-url message]
  (->> (SendMessageRequest. queue-url message)
       (.sendMessageAsync client)
       thread))

(defn send-messages [client queue-url & messages]
  (->> (map-indexed #(SendMessageBatchRequestEntry. (str %1) %2) messages)
       (SendMessageBatchRequest. queue-url)
       (.sendMessageBatchAsync client)
       (thread)))

(defn receive-message [client queue-url]
  (->> (ReceiveMessageRequest. queue-url)
       (.receiveMessageAsync client)
       (thread)))

(defn parse-message [^ReceiveMessageResult msg]
  (.getBody msg))

(defn parse-result-future [result-future]
  (map parse-message (.. result-future get getMessages)))


;;; Example time!

(defn echo
  "Sends and receives a message to given queue a number of times."
  [client queue-url message & {:keys [times] :or {times 1}}]
  (->> (repeat times message)
       (apply send-messages client queue-url))
  (receive-message client queue-url))

(def queue "https://sqs.us-east-1.amazonaws.com/some-queue")

(def client (create-client "client-id" "client-secret"))

(defn echo-example [msg]
  (go
    (->> (echo client  queue msg :times 10) <!
         (parse-result-future)
         (mapv println)))
    (print "SHITS ASYNC YO"))