support input read subsets with ::flow/input-filter pred in state

Author: richhickey

src/main/clojure/clojure/core/async/flow.clj

@@ -174,12 +174,18 @@
   Optionally, a returned init state may contain the
   keys ::flow/in-ports and/or ::flow/out-ports. These should be maps
   of cid -> a core.async.channel. The cids must not conflict with the
-  in/out ids. These channels will become part of the read/write set of
-  the process, but are not otherwise visible/resolvable within the
-  flow. Ports are a way to have data enter or exit the flow from
-  outside. Use :transition to coordinate the lifecycle of these
+  in/out ids. These channels will become part of the input/output set
+  of the process, but are not otherwise visible/resolvable within the
+  flow. Ports are a way to allow data to enter or exit the flow from
+  outside of it. Use :transition to coordinate the lifecycle of these
   external channels.
 
+  Optionally, _any_ returned state, whether from :init, :transition
+  or :transform, may contain the key ::flow/input-filter, a predicate
+  of cid. Only inputs (including in-ports) satisfying the predicate
+  will be part of the next channel read set. In the absence of this
+  predicate all inputs are read.
+
   :transition - optional, (state transition) -> state'
 
   transition will be called when the process makes a state transition,

src/main/clojure/clojure/core/async/flow/impl.clj

@@ -241,10 +241,9 @@
                   outs (into (or outs {}) (::flow/out-ports state))
                   io-id (zipmap (concat (vals ins) (vals outs)) (concat (keys ins) (keys outs)))
                   control (::flow/control ins)
-                  ;;TODO rotate/randomize after control per normal alts?
-                  read-chans (vec (-> ins (dissoc ::flow/control) vals))
+                  read-ins (dissoc ins ::flow/control)
                   run
-                  #(loop [status :paused, state state, count 0, read-chans read-chans]
+                  #(loop [status :paused, state state, count 0, read-ins read-ins]
                      (let [pong (fn []
                                   (let [pins (dissoc ins ::flow/control)
                                         pouts (dissoc outs ::flow/error ::flow/report)]
@@ -254,36 +253,43 @@
                                                        :ins (zipmap (keys pins) (map chan->data (vals pins)))
                                                        :outs (zipmap (keys pouts) (map chan->data (vals pouts)))})))
                            handle-command (partial handle-command pid pong)
-                           [nstatus nstate count read-chans]
+                           [nstatus nstate count read-ins]
                            (try
                              (if (= status :paused)
                                (let [nstatus (handle-command status (async/<!! control))
                                      nstate (handle-transition transition status nstatus state)]
-                                 [nstatus nstate count read-chans])
+                                 [nstatus nstate count read-ins])
                                ;;:running
-                               (let [[msg c] (async/alts!! (into [control] read-chans) :priority true)
+                               (let [;;TODO rotate/randomize after control per normal alts?
+                                     read-chans (let [ipred (or (::flow/input-filter state) identity)]
+                                                  (reduce-kv (fn [ret cid chan]
+                                                               (if (ipred cid)
+                                                                 (conj ret chan)
+                                                                 ret))
+                                                             [control] read-ins))
+                                     [msg c] (async/alts!! read-chans :priority true)
                                      cid (io-id c)]
                                  (if (= c control)
                                    (let [nstatus (handle-command status msg)
                                          nstate (handle-transition transition status nstatus state)]
-                                     [nstatus nstate count read-chans])
+                                     [nstatus nstate count read-ins])
                                    (try
                                      (let [[nstate outputs] (transform state cid msg)
                                            [nstatus nstate]
                                            (send-outputs status nstate outputs outs
                                                          resolver control handle-command transition)]
                                        [nstatus nstate (inc count) (if (some? msg)
-                                                                     read-chans
-                                                                     (vec (remove #{c} read-chans)))])
+                                                                     read-ins
+                                                                     (dissoc read-ins cid))])
                                      (catch Throwable ex
                                        (async/>!! (outs ::flow/error)
                                                   #::flow{:pid pid, :status status, :state state,
                                                           :count (inc count), :cid cid, :msg msg :op :step, :ex ex})
-                                       [status state count read-chans])))))
+                                       [status state count read-ins])))))
                              (catch Throwable ex
                                (async/>!! (outs ::flow/error)
                                    #::flow{:pid pid, :status status, :state state, :count (inc count), :ex ex})
-                               [status state count read-chans]))]
+                               [status state count read-ins]))]
                        (when-not (= nstatus :exit) ;;fall out
-                         (recur nstatus nstate (long count) read-chans))))]
+                         (recur nstatus nstate (long count) read-ins))))]
               ((futurize run {:exec exs})))))))