 (ns acyclic.awstools.core
   (:use [amazonica.core]
-        [amazonica.aws.ec2])
+        [amazonica.aws.ec2]
+        acyclic.utils.pinhole)
   (:require [clj-ssh.ssh :as ssh]
+            [taoensso.timbre :as timbre]
             [ clojure.core.async :as async 
              :refer [<! >! <!! >!! timeout chan alt!! go close!]]))
+(timbre/refer-timbre)
 
 (def cred (read-string (slurp "AWS.clj")))
 
 (apply defcredential (map cred [:access-key :secret-key :endpoint]))
 
-;(describe-images :owners ["self"])
-;aws ec2 request-spot-instances --spot-price 0.005 --instance-count 1 --type "one-time" --launch-specification '{"ImageId" : "ami-a048bec8", "InstanceType" : "t1.micro", "Placement" : { "AvailabilityZone" : "us-east-1c"}, "KeyName" : "telekhine", "SecurityGroupIds" : ["sg-149f3c7e"]}'
-; "IamInstanceProfile" : {"Arn" : "arn:aws:iam::633840533036:instance-profile/datomic-aws-peer" }
-
-
-(def my-subnets {"us-east-1c"  "subnet-08eff44e"
-                 "us-east-1a"  "subnet-f290abda"
-                 "us-east-1b"  "subnet-572dd420"})
-
-(defn request-spots [n & {:keys [zone itype price]}]
-  (let [zone   (or zone "us-east-1c")
-        itype  (or itype "t1.micro")
-        price  (str (or price "0.005"))
-        req    [:spot-price 			price
-                :instance-count 		n
-                :type 			"one-time"
-                :launch-specification 
-                {:image-id 			"ami-5cd51634" ;"ami-a048bec8"
-                 :instance-type 		itype
-                 :placement      {:availability-zone      zone}
-                 :key-name			"telekhine"
-                 :security-groups-ids	["sg-78deaf1d"] 
-                 :subnet-id                  (get my-subnets zone)
-                 :iam-instance-profile       {:arn "arn:aws:iam::633840533036:instance-profile/girder-peer"}}]
-        r (apply request-spot-instances req)]
-    (println req)
-    (map :spot-instance-request-id  (:spot-instance-requests r))))
-
+
+
+(def ^{:private true} paths {:zone  [:launch-specification :placement :availability-zone]
+                             :itype  [:launch-specification :instance-type]
+                             :price [:spot-price]
+                             :n     [:instance-count]})
+
+(defmacro go-try [& forms]
+  `(go (try (do ~@forms)
+            (catch Exception e#
+              (do (debug "Caught exception" e#) nil)))))
 
 (defn patience
-  "Invoke (genfn) every msec until predicate returns true or timeout channel is closed."
-  [pred genfn msec tout]
-  (let [c    (chan)]
-    (close! (async/go-loop []
-               (let [r (genfn)]
-                 (if (or (pred r) (not (first (async/alts! [tout] :default true))))
-                   (do (println "Returning" r)
-                       (>! c r) (close! c))
-                   (do 
-                     (println "Got" r "...waiting" msec)
-                     (<! (timeout msec))
-                     (recur))))))
-    c))
-
-(defn patience-every [pred collfn msec tout]
-  (patience #(every? pred %) collfn msec tout))
+  "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])]
+        (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)))
+                                          (recur)))))
+    out))
+
+
 
 (defn pluck [c]
   (let [a (atom nil)]

                  (reset! a x))))
    a))

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

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

(defn dns-names [is]
  (loop [n 5]
    (let [ds    (:reservations (describe-instances :instance-ids is))
          names (map #(get-in % [:instances 0 :public-dns-name]) ds)]

        :else
        (do
          (Thread/sleep 10000)
          (recur (dec n)))))))

#_(defn request-spots [n & {:keys [zone itype price image key profile group subnet]}]
  (let [zone   (or zone "us-east-1c")
        itype  (or itype "t1.micro")
        price  (str (or price "0.005"))
        image  (or image "BLEH")
        sgs    (if (seq sg) sg [sg])
        req    [:spot-price 			price
                :instance-count 		n
                :type 			"one-time"
                :launch-specification 
                {:image-id 			image
                 :instance-type 		itype
                 :placement {:availability-zone      zone}
                 :key-name			key
                 :security-groups-ids	     sgs
                 :subnet-id                  (get my-subnets zone)
                 :iam-instance-profile       {:arn "arn:aws:iam::633840533036:instance-profile/girder-peer"}}]
        (apply request-spot-instances req)]
    (println req)
    (map :spot-instance-request-id  (:spot-instance-requests r))))

#_(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."
  [pred genfn time-limit time-wait]
  (let [t1     (if (number? time-limit) (timeout (* 1000 time-limit)) time-limit)
        out    (chan)]
    (async/go-loop []
      (let [r   (genfn)
            p   (pred r)]
        (if p
         (do (debug "Returning" r)
              (>! out r) (close! r))
         (let [t2     (timeout (* 1000  time-wait))
               _      (debug "Still" r "Waiting" time-wait "sec")
               [v c] (async/alts! [t1 t2])]
           (if (= c t1)
             (do 
               (debug time-limit "seconds expired.  Returning nil.")
               (close! out))
             (recur))))))
    out))


#_(defn patience-every [pred collfn time-limit time-wait]
  (patience #(every? pred %) collfn time-limit time-wait))


#_(defn- add-sessions [res]
  (let [out   (chan)
        tout  (timeout 60000)
        rs    (:rs res)
        hosts (:hosts res)
        sess  (map #(ssh/session ag % {:strict-host-key-checking :no
                                                  :username "ec2-user"})
                   hosts)]
    (async/go-loop []
      (if (every? ssh/connected? sess)
        (do (debug "All connected.  Returning.")
            (>! out (assoc res :sessions sess)))
        (let 
          (if (= c tout)
            (do
              (debug "Failed to establish connections after timeout.  Terminating.")
              (cancel rs)
              (close! out))
            (if #(every? ssh/connected? sess)
              (do (debug "Returning connected" sess)
                  (>! out (assoc res :sessions sess)))
              (do
                (debug "Waiting 10 sec")
                (<! (timeout 10000))
                (recur)))))))
    out))


#_(defn- add-requests [res]
  (let [req (:req res)
        out (chan)]
    (go 
      (let [rs  (request-spots req)]
        (debug "waiting for" rs "to be active")
        (<! (timeout 1000))
        (if (<! (patience-every #(= "active" %) #(request-status rs) 600 60))
          (do (debug "All instances active! Returning" (assoc res :rs rs))
              (>! out (assoc res :rs rs)))
          (give-up-and-cancel res out "Requests failed.")
)))
    out))
