           (give-up-and-cancel res out "Requests failed.")
 )))
     out))
+
+
+#_(defn request-info [ids]
+  (let [d      (cmd-timeout 5 describe-spot-instance-requests :spot-instance-request-ids (filter #(re-find #"^sir-" %) ids))
+        sirs   (:spot-instance-requests d)
+        rs     (map :spot-instance-request-id sirs)
+        rs->rs (into {} (map #(vector %1 %2) rs0 rs))
+        ss     (map :state sirs)
+        is     (map :instance-id sirs)
+        rs->i  (into {} (map #(vector %1 %2) rs is))
+        is-up  (filter identity is)
+        is-r   (:reservations (describe-instances :instance-ids is-up)) 
+        is-i   (map #(get-in % [:instances 0]) is-r)
+        ;ip-pub (map :public-ip-address is-i)
+        is-up  (map :instance-id is-i)
+        ip-pri (map :private-ip-address is-i)
+        hosts  (map :public-dns-name is-i)
+        hosts  (map #(when (seq %) %) hosts)
+        i->inf (into {} (map #(vector %1 {:instance %1 :ip %2 :host %3}) is-up ip-pri hosts))
+        ]
+    (map #(merge {:request %1 :status %2} ((comp i->inf rs->i) %1)) rs0 ss)))
+
+
+#_(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- add-sessions [req]
+  (let [sess (map #(ssh/session ag % {:strict-host-key-checking :no
+                                      :username "ec2-user"})
+                  (:hosts req))]
+    (patience
+     (fn [] (let [sbad (filter #(not (ssh/connected? %)) sess)]
+              (doseq [s sbad] (ssh/connect s))
+              (assoc req :sessions sess)))
+     #(every? ssh/connected? (:sessions %1))
+     nil ;#(give-up-and-cancel req "Failed to add sesions")
+     60 10
+     )))
+
+#_(defn chain-some
+  "Each cfn must be a function of one parameter, returning a channel that will deliver
+its non-nil result, or nil on failure.  Non-nil return values will be progressively passed
+to each cfn in order.  Returns a channel that will contain the result of the final
+cfn or nil, if one of them failed along the way."
+  [x & cfns]
+  (let [out (chan)]
+    (debug "Starting" cfns)
+    (async/go-loop [[f & fs] cfns
+                    x        x]
+      (if-not f
+        (do (debug "Finished" cfns)
+            (>! out x))
+        (let [_ (debug "Invoking" f x)
+              y (<! (f x))]
+          (if-not y
+            (do (debug "Failed at" f)
+                (close! out))
+            (recur fs y)))))
+    out))
+
+
+;; 
+#_(defn bring-up-aws
+  "Bring up n AWS t1.micro instances and return channel that will contain a single map
+of {:instance ids :hosts names and :sessions objects}, with appropriate timeouts.
+Returns a channel that will contain a map of :hosts, :sessions, etc or nil."
+  [reqmap nmin & opts]
+  (let [req (apply ph-assoc reqmap paths opts)
+        req (assoc req :nmin nmin)
+        n    (ph-get reqmap paths :n)
+        req (if (<= nmin n) req (ph-assoc req paths :n nmin))]
+    (debug "Bringing up" req)
+    (chain-some {:req req} add-requests; add-instances add-ips add-hosts
+                )
+    ))
+;; add-sessions
+
+
+
+#_(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- give-up-and-cancel [req & [msg]]
+  (when msg (debug msg))
+  (concat  (some-> req :is terminate)
+           (some-> req :rs cancel)))
+

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

  (let [d (describe-spot-instance-requests :spot-instance-request-ids rs)]
    (map :instance-id (:spot-instance-requests d))))

(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 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 {}))

