Merge branch 'master' into dev-io-thread

Author: fogus

.github/workflows/generate-codox-docs.yml

@@ -0,0 +1,48 @@
+name: Build Codox Docs
+
+on:
+  workflow_dispatch:
+
+jobs:
+
+  build:
+    runs-on: ubuntu-latest
+    steps:
+
+      - name: Set up Java
+        uses: actions/setup-java@v3
+        with:
+          java-version: 8
+          distribution: 'temurin'
+
+      - name: Set up Clojure
+        uses: DeLaGuardo/[email protected]
+        with:
+          cli: 'latest'
+
+      - name: Cache clojure dependencies
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.m2/repository
+            ~/.gitlibs
+          key: cljdeps-${{ hashFiles('deps.edn') }}
+          restore-keys: cljdeps-
+
+      - name: Clone the repo
+        uses: actions/checkout@v4
+
+      - name: Install rlwrap
+        run: sudo apt-get install -y rlwrap
+
+      - name: Execute doc build
+        run:  |
+          clj -X:docs
+
+      - name: Commit and push
+        run: |
+          git config --global user.name clojure-build
+          git config --global user.email "[email protected]"
+          git add -u -v
+          git commit -m "Action doc commit"
+          git push origin master

docs/clojure.core.async.flow.html

@@ -81,7 +81,7 @@
 channel returned from start</pre></div></div><div class="public anchor" id="var-process"><h3>process</h3><div class="usage"><code>(process process-impl-map)</code><code>(process process-impl-map {:keys [workload timeout-ms], :or {timeout-ms 5000}, :as opts})</code></div><div class="doc"><pre class="plaintext">Given a map of functions (described below), returns a launcher that
 creates a process compliant with the process protocol (see the
 spi/ProcLauncher doc). The possible entries for process-impl-map
-are :describe, :init, :transition, :transform and :introduce. This is
+are :describe, :init, :transition and :transform. This is
 the core facility for defining the logic for processes via ordinary
 functions.
 
@@ -107,9 +107,24 @@
 
 :init - optional, (arg-map) -&gt; initial-state
 
-init will be called once by the process to establish any
-initial state. The arg-map will be a map of param-&gt;val, as supplied
-in the flow def. init must be provided if 'describe' returns :params.
+init will be called once by the process to establish any initial
+state. The arg-map will be a map of param-&gt;val, as supplied in the
+flow def. init must be provided if 'describe' returns :params.
+
+Optionally, a returned init state may contain the
+keys ::flow/in-ports and/or ::flow/out-ports. These should be maps
+of cid -&gt; a core.async.channel. The cids must not conflict with the
+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) -&gt; state'
 
@@ -121,9 +136,7 @@
 process will no longer be used following that. See the SPI for
 details. state' will be the state supplied to subsequent calls.
 
-Exactly one of either :transform or :introduce are required.
-
-:transform - (state in-name msg) -&gt; [state' output]
+:transform - required, (state in-name msg) -&gt; [state' output]
 where output is a map of outid-&gt;[msgs*]
 
 The transform fn will be called every time a message arrives at any
@@ -134,21 +147,6 @@
 may never be nil (per core.async channels). state' will be the state
 supplied to subsequent calls.
 
-:introduce - (state) -&gt; [state' output]
-where output is a map of outid-&gt;[msgs*], per :transform
-
-The introduce fn is used for sources - proc-impls that introduce new data
-into the flow by doing I/O with something external to the flow and
-feeding that data to its outputs. A proc-impl specifying :introduce may not
-specify any :ins in its descriptor, as none but the ::flow/control channel
-will be read. Instead, introduce will be called every time through the
-process loop, and will presumably do blocking or paced I/O to get
-new data to return via its outputs. If it does blocking I/O it
-should do so with a timeout so it can regularly return to the
-process loop which can then look for control messages - it's fine
-for introduce to return with no output. Do not spin poll in the introduce
-fn.
-
 process accepts an option map with keys:
 :workload - one of :mixed, :io or :compute
 :compute-timeout-ms - if :workload is :compute, this timeout (default 5000 msec)
@@ -158,14 +156,11 @@
 any :workload returned by the :describe fn of the process. If neither
 are provded the default is :mixed.
 
-The :compute workload is not allowed for proc impls that
-provide :introduce (as I/O is presumed).
-
 In the :workload context of :mixed or :io, this dictates the type of
 thread in which the process loop will run, _including its calls to
-transform/introduce_. 
+transform_. 
 
-When :io is specified transform/introduce should not do extensive computation.
+When :io is specified, transform should not do extensive computation.
 
 When :compute is specified (only allowed for :transform), each call
 to transform will be run in a separate thread. The process loop will
@@ -182,7 +177,7 @@
 
 :report-chan - a core.async chan for reading.'ping' reponses
 will show up here, as will any explicit ::flow/report outputs
-from :transform/:introduce
+from :transform
 
 :error-chan - a core.async chan for reading. Any (and only)
 exceptions thrown anywhere on any thread inside a flow will appear

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

@@ -86,7 +86,7 @@
   
   :report-chan - a core.async chan for reading.'ping' reponses
   will show up here, as will any explicit ::flow/report outputs
-  from :transform/:introduce
+  from :transform
   
   :error-chan - a core.async chan for reading. Any (and only)
   exceptions thrown anywhere on any thread inside a flow will appear
@@ -141,7 +141,7 @@
   "Given a map of functions (described below), returns a launcher that
   creates a process compliant with the process protocol (see the
   spi/ProcLauncher doc). The possible entries for process-impl-map
-  are :describe, :init, :transition, :transform and :introduce. This is
+  are :describe, :init, :transition and :transform. This is
   the core facility for defining the logic for processes via ordinary
   functions.
 
@@ -167,9 +167,24 @@
 
   :init - optional, (arg-map) -> initial-state
   
-  init will be called once by the process to establish any
-  initial state. The arg-map will be a map of param->val, as supplied
-  in the flow def. init must be provided if 'describe' returns :params.
+  init will be called once by the process to establish any initial
+  state. The arg-map will be a map of param->val, as supplied in the
+  flow def. init must be provided if 'describe' returns :params.
+
+  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 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'
 
@@ -181,9 +196,7 @@
   process will no longer be used following that. See the SPI for
   details. state' will be the state supplied to subsequent calls.
 
-  Exactly one of either :transform or :introduce are required.
-
-  :transform - (state in-name msg) -> [state' output]
+  :transform - required, (state in-name msg) -> [state' output]
   where output is a map of outid->[msgs*]
 
   The transform fn will be called every time a message arrives at any
@@ -194,21 +207,6 @@
   may never be nil (per core.async channels). state' will be the state
   supplied to subsequent calls.
 
-  :introduce - (state) -> [state' output]
-  where output is a map of outid->[msgs*], per :transform
- 
-  The introduce fn is used for sources - proc-impls that introduce new data
-  into the flow by doing I/O with something external to the flow and
-  feeding that data to its outputs. A proc-impl specifying :introduce may not
-  specify any :ins in its descriptor, as none but the ::flow/control channel
-  will be read. Instead, introduce will be called every time through the
-  process loop, and will presumably do blocking or paced I/O to get
-  new data to return via its outputs. If it does blocking I/O it
-  should do so with a timeout so it can regularly return to the
-  process loop which can then look for control messages - it's fine
-  for introduce to return with no output. Do not spin poll in the introduce
-  fn.
-
   process accepts an option map with keys:
   :workload - one of :mixed, :io or :compute
   :compute-timeout-ms - if :workload is :compute, this timeout (default 5000 msec)
@@ -218,14 +216,11 @@
   any :workload returned by the :describe fn of the process. If neither
   are provded the default is :mixed.
 
-  The :compute workload is not allowed for proc impls that
-  provide :introduce (as I/O is presumed).
-
   In the :workload context of :mixed or :io, this dictates the type of
   thread in which the process loop will run, _including its calls to
-  transform/introduce_. 
+  transform_. 
 
-  When :io is specified transform/introduce should not do extensive computation.
+  When :io is specified, transform should not do extensive computation.
 
   When :compute is specified (only allowed for :transform), each call
   to transform will be run in a separate thread. The process loop will
@@ -281,8 +276,7 @@
   and one output (named :in and :out), and no state."
   [f]
   (fn
-    ([] {:params {}
-         :ins {:in (str "the argument to " f)}
+    ([] {:ins {:in (str "the argument to " f)}
          :outs {:out (str "the return of " f)}})
     ([_] nil)
     ([_ _ msg] [nil {:out (f msg)}])))

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

@@ -187,7 +187,10 @@
   "when transition, returns maybe new state"
   [transition status nstatus state]
   (if (and transition (not= status nstatus))
-    (transition state nstatus)
+    (transition state (case nstatus
+                            :exit ::flow/stop
+                            :running ::flow/resume
+                            :paused ::flow/pause))
     state))
 
 (defn send-outputs [status state outputs outs resolver control handle-command transition]
@@ -214,13 +217,11 @@
 
 (defn proc
   "see lib ns for docs"
-  [{:keys [describe init transition transform introduce] :as impl} {:keys [workload compute-timeout-ms]}]
-  (assert (= 1 (count (keep identity [transform introduce]))) "must provide exactly one of :transform or :introduce") 
+  [{:keys [describe init transition transform] :as impl} {:keys [workload compute-timeout-ms]}]
+  (assert transform "must provide :transform") 
   (let [{:keys [params ins] :as desc} (describe)
         workload (or workload (:workload desc) :mixed)]
-    (assert (not (and ins introduce)) "can't specify :ins when :introduce")
     (assert (or (not params) init) "must have :init if :params")
-    (assert (not (and introduce (= workload :compute))) "can't specify :introduce and :compute")
     (reify
      clojure.core.protocols/Datafiable
      (datafy [_]
@@ -229,17 +230,20 @@
      spi/ProcLauncher
      (describe [_] desc)
      (start [_ {:keys [pid args ins outs resolver]}]
+            (assert (or (not params) args) "must provide :args if :params")
             (let [comp? (= workload :compute)
                   transform (cond-> transform (= workload :compute)
                                     #(.get (futurize transform {:exec (spi/get-exec resolver :compute)})
                                            compute-timeout-ms TimeUnit/MILLISECONDS))
                   exs (spi/get-exec resolver (if (= workload :mixed) :mixed :io))
+                  state (when init (init args))
+                  ins (into (or ins {}) (::flow/in-ports state))
+                  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 (into [control] (-> ins (dissoc ::flow/control) vals))
+                  read-ins (dissoc ins ::flow/control)
                   run
-                  #(loop [status :paused, state (when init (init args)), count 0]
+                  #(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)]
@@ -249,40 +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]
+                           [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])
+                                 [nstatus nstate count read-ins])
                                ;;:running
-                               (let [[msg c] (if transform
-                                               (async/alts!! read-chans :priority true)
-                                               ;;introduce
-                                               (when-let [msg (async/poll! control)]
-                                                 [msg control]))
+                               (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])
+                                     [nstatus nstate count read-ins])
                                    (try
-                                     (let [[nstate outputs] (if transform
-                                                              (transform state cid msg)
-                                                              (introduce state))
+                                     (let [[nstate outputs] (transform state cid msg)
                                            [nstatus nstate]
                                            (send-outputs status nstate outputs outs
                                                          resolver control handle-command transition)]
-                                       [nstatus nstate (inc count)])
+                                       [nstatus nstate (inc count) (if (some? msg)
+                                                                     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])))))
+                                       [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]))]
+                                   #::flow{:pid pid, :status status, :state state, :count (inc count), :ex ex})
+                               [status state count read-ins]))]
                        (when-not (= nstatus :exit) ;;fall out
-                         (recur nstatus nstate (long count)))))]
+                         (recur nstatus nstate (long count) read-ins))))]
               ((futurize run {:exec exs})))))))