 (defn healthy-backend?
   "true if backend seen more recently than time"
   [time b]
-  (> (s->millepoch-time (:last-seen-at b))
-     time))
+  (> (:last-seen-at-millepoch b) time))
 
-(defn run-server [prefix]
-  (let [shutdown-ch (chan)
-        config-watcher (etcd/watch-prefix prefix)
-        backends-ch (chan)]
+(defn combined-config-watcher [prefix]
+  (let [config-watcher (etcd/watch-prefix prefix)
+        out-ch (chan)]
     (go
-      (loop [listeners {}
-             backends {}
-             backend-prefix nil
+      (loop [backend-prefix nil
              backend-watcher nil]
         (let [config (etcd/get-prefix prefix)
               new-bp (:upstream-service-etcd-prefix config)
-              backends (if new-bp
-                         (update-backends-from-config
-                          backends (etcd/get-prefix new-bp))
-                         backends)
-              listeners (update-listeners-from-config
-                         backends-ch
-                         listeners
-                         (:listen-address config))]
-          (print-state listeners backends)
-          (let [[value ch]
-                (alts! (remove nil?
-                               [config-watcher
-                                backend-watcher
-                                shutdown-ch
-                                backends-ch]))]
-            (when (= ch backends-ch)
-              (let [since (- (millepoch-time-now)
-                             (* 1000 (Integer/parseInt (:upstream-freshness config))))
-                    good-backends (filter (partial healthy-backend? since)
-                                          (vals backends))]
-                (if (seq good-backends)
-                  (alts! (map (fn [b] [(:channel b) value]) good-backends))
-                  (>! (second value) {:status 500 :headers {}
-                                      :body "No healthy downstreams"}))))
-            (if (= ch shutdown-ch)
-              (.close (first (vals listeners)))
-              (recur listeners
-                     backends
-                     new-bp
-                     (refresh-backend-watcher backend-watcher backend-prefix new-bp))))))
-      (println "quit"))
+              backend-watcher
+              (if (and backend-watcher (= new-bp backend-prefix))
+                backend-watcher
+                (and new-bp (etcd/watch-prefix new-bp)))]
+          (>! out-ch {:config config
+                      :backends (if new-bp (etcd/get-prefix new-bp) {})})
+          (alts! (remove nil? [config-watcher backend-watcher]))
+          (recur new-bp backend-watcher))))
+    out-ch))
+
+(defn update-backends-state [app-state backends-spec]
+  (let [before (reduce (fn [m bk]
+                         (assoc m (:listen-address bk) bk))
+                       {}
+                       (:backends app-state))
+        after  (reduce (fn [m bk]
+                         (assoc m (:listen-address bk) bk))
+                       {}
+                       (vals backends-spec))
+        b-addrs (set (keys before))
+        a-addrs (set (keys after))
+        deleted (set/difference b-addrs a-addrs)
+        create (set/difference a-addrs b-addrs)]
+    (pprint [:update-backends before after deleted create])
+    (dorun (map (fn [b] (close! (:channel (get before b)))) deleted))
+    (map (fn [a]
+           (let [bk (get after a)]
+             (assoc bk
+                    :channel
+                    (or (:channel (get before a))
+                        (make-backend-channel bk))
+                    :last-seen-at-millepoch
+                    (s->millepoch-time (:last-seen-at bk))
+                    )))
+         a-addrs)))
+
+(defn open-listener [addr dispatch-ch]
+  (http/start-server (partial proxy-handler dispatch-ch)
+                     (parse-address addr)))
+
+(defn update-app-state [app-state m]
+  (println [:update-app-state app-state m])
+  (let [dispatch-ch (:dispatch-channel app-state)
+        config (:config m)
+        backends-spec (:backends m)
+        addr (:listen-address config)]
+    (assoc app-state
+           :upstream-freshness-ms
+           (* 1000 (Integer/parseInt (:upstream-freshness config)))
+           :dispatch-channel dispatch-ch
+           :listen-address (:listen-address config)
+           :listener
+           (if-let [l (:listener app-state)]
+             (if (= (:listen-address app-state) addr)
+               l
+               (do
+                 (.close l)
+                 (open-listener addr dispatch-ch)))
+             (open-listener addr dispatch-ch))
+           :backends (update-backends-state app-state backends-spec))))
+
+(defn run-server [prefix]
+  (let [config-ch (combined-config-watcher prefix)
+        shutdown-ch (chan)
+        dispatcher-ch (chan)]
+    (go
+      (loop [app-state {:dispatch-channel dispatcher-ch}]
+        (let [backends (:backends app-state)
+              [val ch] (alts! [config-ch dispatcher-ch shutdown-ch])]
+          (condp = ch
+            config-ch
+            (recur (spy (update-app-state app-state val)))
+            dispatcher-ch
+            (let [since (- (millepoch-time-now)
+                           (:upstream-freshness-ms app-state))
+                  good-backends (filter (partial healthy-backend? since)
+                                        backends)]
+              (if (seq good-backends)
+                (alts! (map (fn [b] [(:channel b) val]) good-backends))
+                (>! (second val) {:status 500 :headers {}
+                                  :body "No healthy downstreams"}))
+              (recur app-state))
+            shutdown-ch
+            (do
+              (dorun (map (fn [b] (async/close! (:channel b))) backends))
+              (when-let [l (:listener app-state)]
+                (.close l))
+              (println "stopping"))))))
     shutdown-ch))
-
-
-
-
 
 (defn -main
   "I don't do a whole lot ... yet."
   [& args]
   (println "Hello, World!")
-;  (add-watch etcd/configuration :update-frontend update-frontend)
-;  (etcd/watch-config (first args))
-  (println "running"))
+  (let [s (run-server (first args))]
+    (Thread/sleep (* (Integer/parseInt (second args)) 1000))
+    (async/put! s :done)))

            [aleph.http :as http])
  (:gen-class))

(defn refresh-backend-watcher [watcher old-prefix new-prefix]
  (if (and watcher (= old-prefix new-prefix))
    watcher
    (do (if watcher (close! watcher))
        (etcd/watch-prefix new-prefix))))
(defn spy [a] (pprint a) a)

(defn proxy-handler1 [backends-ch req]

      (println [:shutdown backend]))
    ch))

(defn s->millepoch-time [s]
  (.getTimeInMillis (clojure.instant/read-instant-calendar s)))

