removed :introduce, use ports instead for sources/sinks, added ::flow/[in-out-]ports support for external I/O, fix transitions

Author: richhickey

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,18 @@
 
   :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 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
+  external channels.
 
   :transition - optional, (state transition) -> state'
 
@@ -181,9 +190,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 +201,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 +210,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 +270,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,21 @@
      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))
                   run
-                  #(loop [status :paused, state (when init (init args)), count 0]
+                  #(loop [status :paused, state state, count 0]
                      (let [pong (fn []
                                   (let [pins (dissoc ins ::flow/control)
                                         pouts (dissoc outs ::flow/error ::flow/report)]
@@ -256,20 +261,14 @@
                                      nstate (handle-transition transition status nstatus state)]
                                  [nstatus nstate count])
                                ;;:running
-                               (let [[msg c] (if transform
-                                               (async/alts!! read-chans :priority true)
-                                               ;;introduce
-                                               (when-let [msg (async/poll! control)]
-                                                 [msg control]))
+                               (let [[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])
                                    (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)]