   (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))]


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

;(create-queue (uuid))

(defn queue-reduce [f xs in-queue out-queue]
  (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 1024 xs) (range))]
    (thread
      (doseq [[batch index] batches]
        (send-message in-queue {:type "process" :index index :payload batch})))
    (let [response (reduce (fn [acc _]
                             (f acc (:payload (<!! results))))
                           0
                           batches)]
      (send-message in-queue {:type "stop"})
      (stop-fn)
      response)))

;(comment
(defn do-reduce []
  (let [in (create-queue (uuid))
        out (create-queue (uuid))]
    (let [response (queue-reduce + (range 10000) in out)]
      (Thread/sleep 200)
      (delete-queue in)
      (delete-queue out)
      response)))
  ;(super-reduce plus (range 100000))



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


(defn send-ok-messages [c buffer]
  (if-let [result (aq/peek-head buffer)]
    (do
      (go (>! c result))
      (send-ok-messages c (aq/remove-one buffer)))
    buffer))
