 
 
 
-(defn connect [peer conn-ch]
+(defn connect [peer conn-ch out]
   (go-loop [{:keys [ip4 port] :as c} (<! conn-ch)]
            (when c
              (let [[bus-in bus-out] (:chans (:volatile @peer))

            (<! (wire peer [in (async/pub out :type)]))
            (assoc stage :chans [(async/pub in :type) out])))))

;; push perspective
(defn sync! [{:keys [type author chans new-values meta] :as stage}]
  (go (let [[p out] chans

#_(clojure.pprint/pprint @(:log (:volatile @peer-a)))
#_(clojure.pprint/pprint @stage-log)
#_(let [in (debug/chan stage-log [:stage :in])
      out (debug/chan stage-log [:stage :out])
      b-in (debug/chan stage-log [:peer-b :in])
      b-out (debug/chan stage-log [:peer-b :out])]
    (go-loop [m (<! b-in)]
                (println "PEERB-IN" m)
                (recur (<! b-in)))
  (go (<! (wire peer [in (async/pub out :type)]))
      (<! (wire peer-b [b-in (async/pub b-out :type)]))
      (>! b-out {:type :meta-sub :metas {"john" #{1}}})
      (<! (timeout 1000))
      (>! b-out {:type :connect :ip4 "127.0.0.1" :port 9090})

      (>! out {:type :meta-sub :metas {"john" #{1}}})
;      (<! in)
      (<! (timeout 1000))
      (>! out {:type :connect
               :ip4 "127.0.0.1"
               :port 9090})
      #_(>! out {:type :connect
               :ip4 "127.0.0.1"
               :port 9091})
      (<! (timeout 1000))
      (>! out {:type :meta-pub
               :user "john" :meta {:id 1
