 
 (defn handle-message [state]
   (fn [msg]
-    (info "got message:" msg)
     (a/go
       (a/>! (:rx-chan @state) msg))))
 

  true)

(defn handle-rx-pkt [state dt pkt]
  (warn "handle-rx-pkt")
  (swap! state (fn [s]
                 (-> s
                     (update-in [:rx-queue] conj pkt)

(defn incoming-message-loop
  [f state]
  (fn [s info]
    (d/let-flow [handshake (s/take! s)
                 _ (s/put! s (:id @state))]
                (if (get-in @state [:peer-sockets handshake])
                  (warn (:id @state )": already connected to: "  handshake)
                  (swap! state assoc-in [:peer-sockets handshake] s))
                (d/loop []
                  (warn "loop..." (:id @state))
                  (-> (s/take! s ::none)
                      (d/chain
                       (fn [msg]
                         (if (= ::none msg)
                           (do
                             (warn "closing!")
                             (s/close! s))
                           (do
                             (f msg)
                             (d/recur)))))
                      (d/catch
                          (fn [ex]
                            (warn ex "in close")
                            (s/put! s (str "CLOSED:" ex))
                            (s/close! s))))))))

(defn now []
  (System/currentTimeMillis))
