 (ns distributed-transducers-poc.rc
   (:require [cheshire.core :refer [generate-string parse-string parse-stream]]
-            [clojure.core.async :refer [go]]
+            [clojure.core.async :refer [chan go go-loop >! <! <!!]]
             [clojure.java.io :as io]
             [serializable.fn :as s])
   (:import [com.amazonaws ClientConfiguration]

          response {:result (f (:param payload))}]
      (send-message out-queue response))))

(defn uuid [] (str (java.util.UUID/randomUUID)))

(defn queue-reduce [f xs in-queue out-queue]
  (let [batches (partition-all 1024 xs)]
    (reduce (fn [acc batch]
              (send-message in-queue {:param batch})
              (message-handler #(reduce f %) in-queue out-queue)
              (let [response (receive-message out-queue 5)]
                (f acc (:result response))))
            0
            batches)))

(comment
(let [in (create-queue (uuid))

        (.getBody)
        (to-json))))

(defn receive-messages [queue-url wait-in-seconds]
  (let [to-json #(parse-string % true)
        raw-messages (-> (.receiveMessage sqs-client (-> (ReceiveMessageRequest.)
                                        (.withQueueUrl queue-url)
                                        (.withWaitTimeSeconds (int wait-in-seconds))))
        (.getMessages))]
    (map #(-> % .getBody to-json) raw-messages)))

(defn delete-queue [queue-url]
  (.deleteQueue sqs-client queue-url))

