       (reset! peer @(client-peer "CLIENT" (new-store)))
       (reset! stage-log {}))
 #_(clojure.pprint/pprint @(:log (:volatile @peer)))
-#_(clojure.pprint/pprint @(:log (:volatile @peer-a)))
+#_(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 :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
+  (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 {:type :connect
-               :ip4 "127.0.0.1"
-               :port 9091})
-      (<! (timeout 1000))
-      (>! out {:type :meta-pub
-               :user "john" :meta {:id 1
-                                   :causal-order {1 #{}
-                                                  2 #{1}}
-                                   :last-update (java.util.Date. 0)
-                                   :schema {:type ::geschichte
-                                            :version 1}}})
-      (<! (timeout 1000))
-;     (<! in)
-      (>! out {:type :fetched :values {1 2
-                                       2 42}})
-      (<! (timeout 1000))
-;     (println "1" (:type (<! in)))
-;     (println "2" (:type (<! in)))
-      (>! out {:type :meta-pub
-               :user "john" :meta {:id 1
-                                   :causal-order {1 #{}
-                                                  2 #{1}
-                                                  3 #{2}}
-                                   :last-update (java.util.Date. 0)
-                                   :schema {:type ::geschichte
-                                            :version 1}}})
-      (<! (timeout 1000))
-      (>! out {:type :fetched :values {3 43}})
-;     (println "4" (:type (<! in)))
-;     (println "5" (:type (<! in)))
-      (<! (timeout 5000)))
+      #_(>! 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)
+           (println "RECEIVED:" i)
            (recur (<! in))))
 
 

                                       "Testing."
                                       false
                                       {:some 42})
            stage (<! (wire-stage peer-b stage))
            [in out]  (:chans stage)
            connedch (chan)
            _ (async/sub in :connected connedch)
            _ (go-loop [conned (<! connedch)]
                       (println "CONNECTED-TO:" conned)
                       (recur (<! connedch)))
;           _ (>! out {:type :meta-sub
;                      :metas {"me@mail.com" #{(:id (:meta stage))}}})
;           _ (<! (timeout 1000))
            _ (>! out {:type :connect
                       :ip4 "127.0.0.1"
                       :port 9090})
            _ (<! (timeout 1000))
;            new-stage (<! (sync! stage))
            ]
        #_(-> new-stage
            (s/transact {:helo :world} '(fn more [old params] old))
            repo/commit
            sync!

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

