                     :ip ip
                     :port port})]
     (go-loop [[in out] (<! new-conns)]
-             (wire peer [out (async/pub in :type)])
+             (wire peer [out (async/pub in :topic)])
              (recur (<! new-conns)))
     peer))
 

               (client-connect! ip4 port c-in c-out)

                                        ; handshake
               (>! c-out {:type :meta-sub :metas subs})
               (put! subed-ch (<! c-in))
               (subscribe peer subed-ch bus-out c-out)

               (<! (wire peer [c-out (async/pub c-in :type)]))
               (>! out {:type :connected
                        :ip4 ip4 :port port})
               (recur (<! conn-ch))))))


      (reset! peer @(client-peer "CLIENT" (new-store)))
      (reset! stage-log {}))
#_(clojure.pprint/pprint @(:log (:volatile @peer)))
#_(clojure.pprint/pprint @(:log (:volatile @peer-b)))
#_(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 :topic)]))
      (<! (wire peer-b [b-in (async/pub b-out :topic)]))
      (>! b-out {:topic :connect :ip4 "127.0.0.1" :port 9090})
      (<! (timeout 100))
      #_(>! b-out {:topic :meta-sub :metas {"john" #{1}}})
      (<! (timeout 100))
      (>! out {:topic :meta-sub :metas {"john" #{1}}})
                                        ;      (<! in)
      (<! (timeout 100))
      (>! out {:topic :connect
               :ip4 "127.0.0.1"
               :port 9090})
      #_(>! out {:topic :connect
                 :ip4 "127.0.0.1"
                 :port 9091})
      (<! (timeout 100))
      (>! out {:topic :meta-pub
               :user "john"
               :meta {:id 1
                      :causal-order {1 #{}
                                     2 #{1}}
                      :last-update (java.util.Date. 0)
                      :schema {:topic ::geschichte
                               :version 1}}})
      (<! (timeout 100))
                                        ;     (<! in)
      (>! out {:topic :fetched :values {1 2
                                        2 42}})
      (<! (timeout 100))
                                        ;     (println "1" (:topic (<! in)))
                                        ;     (println "2" (:topic (<! in)))
      (>! out {:topic :meta-pub
               :user "john"
               :meta {:id 1
                      :causal-order {1 #{}
                                     2 #{1}
                                     3 #{2}}
                      :last-update (java.util.Date. 0)
                      :schema {:topic ::geschichte
                               :version 1}}})
      (<! (timeout 100))
      (>! out {:topic :fetched :values {3 43}})
                                        ;     (println "4" (:topic (<! in)))
                                        ;     (println "5" (:topic (<! in)))
      (<! (timeout 500)))


  (go-loop [i (<! in)]
           (println "RECEIVED:" i)
           (recur (<! in))))


