   (let [d (describe-spot-instance-requests :spot-instance-request-ids rs)]
     (map :instance-id (:spot-instance-requests d))))
 
-(defn cancel [rs]
-  (let [c (:cancelled-spot-instance-requests
-        (cancel-spot-instance-requests :spot-instance-request-ids rs))]
-    (debug c)
-    c))
+(defn request-spots [req & opts]
+  (let [req    (apply ph-assoc req paths opts)
+        args   (apply concat (seq req))
+        _      (debug "Spot request" (pr-str args))
+        rs     (apply request-spot-instances args)]
+    (map :spot-instance-request-id  (:spot-instance-requests rs))))
 
 (defn terminate [is]
   (let [t (terminate-instances :instance-ids is)]
     (debug t)
     t))
 
-(defn- give-up-and-cancel [req & [msg]]
-  (when msg (debug msg))
-  (concat  (some-> req :is terminate)
-           (some-> req :rs cancel)))
-
-(defn request-spots [req]
-  (let [args    (apply concat (seq req))
-        rs     (apply request-spot-instances args)]
-    (debug "Spot request" args "->" rs)
-    (map :spot-instance-request-id  (:spot-instance-requests rs))))
-
-(defn- add-requests [req]
-  (let [req (assoc req :rs (request-spots (:req req)))]
-    (patience
-     #(assoc req :req-status (cmd-timeout 5 request-status (:rs req)))
-     (fn [req] (let [rss (:req-status req)]
-                 (and (seq rss) (every? #(= "active" %) rss))))
-   #(give-up-and-cancel req "Failed to fulfill spot requests")
-   600 60)))
-
-(defn- add-instances [req]
-  (patience 
-   #(assoc req :is (request-instances (:rs req)))
-   #(good-strings (:is %))
-   #(give-up-and-cancel req "Failed to get instance ids")
-   10 1))
-
-
-(defn- add-hosts [req]
-  (patience
-   (fn [] (assoc req :hosts (->> (:is req)
-                                 (describe-instances :instance-ids)
-                                 :reservations
-                                 (map #(get-in % [:instances 0 :public-dns-name])))))
-   #(good-strings (:hosts %))
-   #(give-up-and-cancel req "Failed to get hosts.")
-   10 1))
-
-
-(defn- add-ips [req]
-  (patience
-   (fn [] (assoc req :ips (->> (:is req)
-                                 (describe-instances :instance-ids)
-                                 :reservations
-                                 (map #(get-in % [:instances 0 :private-ip-address])))))
-   #(good-strings (:ips %))
-   #(give-up-and-cancel req "Failed to get IPs")
-   10 1))
+(defn stop [is]
+  (let [t (stop-instances :instance-ids is)]
+    (debug t)
+    t))
+
+(defn cancel-requests [rs]
+  (let [ds (:spot-instance-requests (describe-spot-instance-requests :spot-instance-request-ids rs))
+        is (filter (complement nil?) (map :instance-id ds))
+        cr (:cancelled-spot-instance-requests (cancel-spot-instance-requests :spot-instance-request-ids rs))
+        ci (and (seq is) (terminate is))]
+    [cr ci]))
+
+
+(def ids->chs (atom {}))
+
+(defn notify-chan [id]
+  (let [c (chan)]
+    (swap! ids->chs assoc id c)
+    c))
+
+(defn close-notify-chan! [id]
+  (swap! ids->chs #(let [c (get % id)]
+                     (close! c)
+                     (dissoc % id)))
+  nil)
+
+
+(defn start-up-listener []
+  (let [ctl  (chan)
+        cl   (sqs-listen)]
+    (swap! ids->chs #(assoc % :sqs-listen cl))
+    (async/go-loop []
+      (let [[v c] (async/alts! [ctl cl])]
+        (cond
+         (= c ctl) (do (reset! ids->chs {}) (close! cl))
+         (= c cl)  (do (try 
+                          (let [s   (b64->ec2-data v)
+                                id  (:id s)
+                                c   (get @ids->chs id)]
+                            (debug "Got" id c s)
+                            (when c (>! c s)))
+                          (catch Exception e (info e)))
+                       (recur)))))
+    ctl))
+
+
+;;aws --region us-east-1 sns publish --topic-arn  arn:aws:sns:us-east-1:633840533036:instance-up --message yowsassl
+
+(defn send-up [id & {topic :topic region :region}]
+  (str
+   "aws --region " (or region my-region)
+   " sns publish --topic-arn " (or topic my-sns-topic)
+   " --message `(echo \"id: " id "\";bin/ec2-metadata -i -p -o) | base64 -w 0`"))
+
+;(stop-instances :instance-ids ["i-7c949193"])
+;(modify-instance-attribute :instance-id "i-7c949193" :user-data (s->b64 "echo hello"))
+
+
+(defn bring-up-spots  [reqmap nmin cmds & opts]
+  (let [cmd (str (send-up) "\n "(clojure.string/join "\n" cmds))
+        req (apply ph-assoc reqmap paths opts)
+        req (ph-assoc req paths :udata cmd)
+        ;req (assoc req :nmin nmin)
+        n   (ph-get reqmap paths :n)
+        req (if (<= nmin n) req (ph-assoc req paths :n nmin))
+        id  (.toString (UUID/randomUUID))
+        _   (debug "Requesting:" id req)
+        cl  (notify-chan id)
+        rs  (request-spots req)
+        to  (timeout (* 5 60 1000))]
+    (debug "Spot requests:" rs)
+    (async/go-loop [i 0]
+      (if (>= i nmin)
+        (let [rs->info (request-info-map rs)
+              rs-up    (filter rs->info rs)
+              rs-dn    (filter (complement rs->info) rs)]
+          (close-notify-chan! id)
+          (cancel-requests rs-dn)
+          (map rs->info rs-up))
+        (let [[v c] (async/alts! [cl to])]
+          (cond
+           (= c to) (do (info "Timeout on spot request:" req)
+                      (close-notify-chan! id)
+                      (cancel-requests rs)
+                      nil)
+           (= c cl) (recur (inc i))))))))
+
 
 (def ag (ssh/ssh-agent {}))
 

(ns acyclic.awstools.core
  (:use [amazonica.core]
        [amazonica.aws.ec2]
        acyclic.utils.pinhole)
  (:require [clj-ssh.ssh :as ssh]
            [taoensso.timbre :as timbre]
            [clojure.data.codec.base64 :as b64]
            [clojure.core.async :as async 
             :refer [<! >! <!! >!! timeout chan alt!! go close!]]))
(timbre/refer-timbre)

; ami-5cd51634



(def cred (read-string (slurp "AWS.clj")))

(apply defcredential (map cred [:access-key :secret-key :endpoint]))

(defn s->b64 [s] (String. (b64/encode (.getBytes s))))


(def ^{:private true} paths {:zone  [:launch-specification :placement :availability-zone]
                             :itype [:launch-specification :instance-type]
                             :price [:spot-price]
                             :n     [:instance-count]
                             :udata [s->b64 :launch-specification :user-data]})


(defmacro go-try [& forms]
  `(go (try (do ~@forms)
            (catch Exception e#
              (do (debug "Caught exception" e#) nil)))))

(defn patience
  "Invoke (genfn) every time-wait sec until predicate returns true or time-limit sec has passed.
Returns result of genfn, or nil in the case of timeout."
  [genfn pred failure time-limit time-wait]
  (let [t1     (timeout (* 1000 time-limit))
        out    (chan)]
    (async/go-loop []
      (let [rc     (go-try (genfn))
            [v c]  (async/alts! [rc t1 out])]
        (debug "Ats:" v c)
        (cond
         (= c out)                    (do (debug "Canceled")
                                          (when failure (failure)))
         (= c t1)                     (do (debug time-limit "sec expired")
                                          (when failure (failure))
                                          (close! out))
         (and (= c rc) v (pred v))    (do (debug "Returning" v)
                                          (>! out v))
         :else                        (do (debug "Still" v "Waiting" time-wait)
                                          (<! (timeout (* 1000 time-wait)))
                                          (debug "Woke up after wait...")
                                          (recur)))))
    out))



(defn pluck [c]
  (let [a (atom nil)]
    (close! (go (let [x (<! c)]
                  (reset! a x))))
    a))


(defn request-spots [req]
  (let [req    (apply concat (seq req))
        _      (debug req)
        rs     (apply request-spot-instances req)]
    (map :spot-instance-request-id  (:spot-instance-requests rs))))

(defn cmd-timeout [sec cmd & args]
  (let [cc (go (apply cmd args))




(defn EDNify
"EDN-ify an arbitrary object, leaving it alone if it's an innocuous string."
  [x] 
