   (vertices [_]))
 
 (defprotocol IGraphExecutor
-  (execute! [_ vertices]))
-
-(defn default-collect
-  [vertex]
-  (swap! (:state vertex) #(update % ::val into (::uncollected %))))
+  (execute! [_])
+  (stop! [_]))
+
+(defn simple-collect
+  [collect-fn]
+  (fn [vertex]
+    (swap! (:state vertex) #(update % ::val collect-fn (::uncollected %)))))
+
+(def default-collect (simple-collect into))
 
 (defn signal-forward
   [state _] (::val state))

                    (>! ctrl [:signal id])
                    (signal! vertex))))
              (recur))
          (info id "stopped"))))
    vertex))

(defn execution-result


(defrecord Vertex [id state last-sig-state in outs]
  clojure.lang.IDeref
  (deref
    [_] (::val @state))
  IVertex
  (collect!
    [_] ((::collect-fn @state) _))
  (score-collect
    [_] ((::score-collect-fn @state) _))
  (connect-to!
    [_ v sig-fn opts]
    (debug (format "%d connecting to %d (%s)" id (:id v) (pr-str opts)))
    (swap! outs assoc v [sig-fn opts]) _)
  (disconnect-vertex!
    [_ v]
    (debug (format "%d disconnecting from %d" id (:id v)))
    (swap! outs dissoc v)
    _)
  (close-input! [_]
    (if @in
      (do (debug (str id " closing..."))
          (close! @in)
          (reset! in nil))
      (warn (str id " already closed")))
    _)
  (input-channel
    [_] (or @in (reset! in (chan))))
  (score-signal
    [_] ((::score-signal-fn @state) _))
  (signal!
    [_]
    (let [state @state]
