   (let [handler (fn [message]
                   (when (= "process" (:type message)
                            (let [result (f (:payload message))]
-                             (println "Result " result)
                              (send-message out-queue {:index (:index message)
                                                       :payload result}))))
                     (:type message))]


(defmacro super-reduce [f xs]
  `(invoke-lambda (pr-str (s/fn [] (reduce ~f ~xs))) "distributed-transducers-poc" "eu-west-1"))

(defn plus [a b]
  (+ a b))

(def sqs-client (AmazonSQSClient. aws-credentials))



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

(defn queue-reduce [f xs in-queues out-queue]
  (doseq [in-queue in-queues]
    (message-loop (fn [x] (reduce f x)) in-queue out-queue))
  (let [{:keys [results stop-fn]} (handler-loop out-queue)
        batches (map (fn [a b] [a b]) (partition-all 4096 xs) (range))]
    (thread
      (doseq [[batch index] batches]
        (send-message (nth in-queues (mod index (count in-queues)))
                      {:type "process" :index index :payload batch})))
    (let [response (reduce (fn [acc _]
                             (f acc (:payload (<!! results))))
                           0
                           batches)]
      (doseq [in-queue in-queues]
        (send-message in-queue {:type "stop"}))
      (stop-fn)
      response)))

(defn uber-reduce [f xs node-count]
  (let [in-queues  (map (fn [_] (create-queue (uuid)))
                        (range node-count))
        out (create-queue (uuid))]
    (let [response (queue-reduce f xs in-queues out)]
      (Thread/sleep 200)
      (doseq [in-queue in-queues]
        (delete-queue in-queue))
      (delete-queue out)
      response)))

;(invoke-lambda (pr-str (s/fn [] (map #(* 2 %) [3 4 5]))) "distributed-transducers-poc" "eu-west-1")

