 (defn wire-stage [peer {:keys [chans] :as stage}]
   (go (if chans stage
           (let [log (atom {})
-                in (debug/chan log ["STAGE" :in])
+                in (debug/chan log ["STAGE" :in] 10)
                 out (debug/chan log ["STAGE" :out])]
-            (<! (wire peer [in (async/pub out :type)]))
-            (assoc stage :chans [(async/pub in :type) out])))))
+            (<! (wire peer [in (async/pub out :topic)]))
+            (assoc stage :chans [(async/pub in :topic) out])))))
 
 
 ;; push perspective

  (let [{:keys [new-conns] :as server} (start-server! ip port)
        log (atom {})
        in (debug/chan log [ip :in])
        out (async/pub in (fn [{:keys [user meta]}] [user (:id meta)]))
        peer (atom {:volatile (merge server
                                     {:store store
                                      :log log

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

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

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

