You signed in with another tab or window.Reload to refresh your session.You signed out in another tab or window.Reload to refresh your session.You switched accounts on another tab or window.Reload to refresh your session.Dismiss alert
Copy file name to clipboardExpand all lines: doc/flow-guide.md
+23-21Lines changed: 23 additions & 21 deletions
Original file line number
Diff line number
Diff line change
@@ -4,15 +4,15 @@
4
4
5
5
The[flow](https://clojure.github.io/core.async/flow.html) library enables a strict separation application logic from the deployment concerns of topology, execution, communication, lifecycle, monitoring and error handling.
6
6
7
-
##Step fns andprocesses
7
+
##Step fns andprocess launchers
8
8
9
9
You provide logic to flow in the form of_step-fns_, which are wrapped into running processes, executing in a loop. Flow manages the life cycle of the process and handles incoming and outgoing messages by putting or taking them on channels. Step-fns do not access channels directly or hold state, making them easy to test in isolation and reuse.
The describe arity must return a static description of the step-fn's:params,:ins, and:outs. Each of these is a map of name (a keyword) to docstring.
18
18
@@ -26,48 +26,48 @@ For example, the describe arity might return this description for a simple step-
26
26
27
27
The names used for input and output channels should be distinct (no overlap).
28
28
29
-
###init (1 arity)
29
+
###init: (step-fn arg-map) -> init-state
30
30
31
31
The init arity is called once by the process to takes a set of args from the flow def (corresponding to the params returned from the describe arity) and returns the init state of the process.
Theprocess state isa map. It can contain any keys needed bythestep-fn transition andtransform arities. In addition, there are some flow-specific keys, described here.
35
+
Thetransition arity iscalled any time the flow or process undergoes a lifecycle transition (::flow/start, ::flow/stop, ::flow/pause, ::flow/resume). The description arity takesthecurrent state andreturns an updated state to be used for subsequent calls.
36
36
37
-
`::flow/pid` is added to thestate bytheprocess based on the name suppliedinthe flow def.
37
+
The step-fn should use thetransition arity to coordinatethecreation, pausing, and shutdown of external resourcesina process.
38
38
39
-
`::flow/in-ports` and`::flow/out-ports` are maps of cid to external channel, optionally returned in the initialstatefrom the init arity. The in-ports andout-ports are used to connect source and sink processes to external channels. These channels must be provided by the step-fn and returned in the init arity map, either by creating the channel or using a channel passed in via the flow def init args for the process. The flow does not manage the lifecycle of these channels.
`::flow/input-filter`, a predicateofcid, can be returned inthestate from any aritytoindicate a filter ontheprocessinputchannel read set. For example, a step-fn that is waiting for a response from multiple inputs might remove the channels that have already responded from the read-set until responses have been received from all.
41
+
The transform arity is called in a loop by the process for every message received on aninput channel and returns a new state and a mapofoutput cids to messages to return. The process will take care of sending these messages totheoutput channels. Output can be senttonone, any or all ofthe:outsenumerated, and/or aninputnamed by a[pid inid] tuple (e.g. for reply-to), and/or to the ::flow/report output. A step need not output at all (output or msgs can be empyt/nil), however an output_message_ may never be nil (per core.async channels).
42
42
43
-
###transition (2arity)
43
+
The step-fn may throw excepitons from anyarity and they will be handled by flow. Exceptions thrown from the transition or transform arities, the exception will be logged on the flow's:error-chan.
44
44
45
-
The transition arity is called any time the flow or process undergoes a lifecycle transition (::flow/start, ::flow/stop, ::flow/pause, ::flow/resume). The description arity takes the currentstate and returns an updated state to be used for subsequent calls.
45
+
###Processstate
46
46
47
-
Thestep-fn should use the transition arity to coordinatethecreation, pausing, andshutdown of external resources in a process.
47
+
Theprocess state is a map. It can contain any keys needed bythestep-fn transition andtransform arities. In addition, there are some flow-specific keys, described here.
48
48
49
-
###transform (3 arity)
49
+
`::flow/pid` is added to the state by the process based on the name supplied in the flow def.
50
50
51
-
The transform arity is called in a loop by the process for every message received on an input channel and returns a new stateanda map of output cidstomessages to return. The process will take care of sending these messages totheoutput channels. Output can be sent to none, any or all ofthe:outsenumerated, and/oran input named by a[pid inid] tuple (e.g. for reply-to), and/or tothe::flow/report output. A step need notoutput at all (output or msgs can be empyt/nil), however an output_message_ may never be nil (per core.asyncchannels).
51
+
`::flow/in-ports` and`::flow/out-ports` are maps of cid to external channel, optionally returned in the initial state from the init arity. The in-portsandout-ports are usedtoconnect source and sink processes to external channels. These channels must be provided bythestep-fn and returned in the init arity map, either by creatingthechannelorusing a channel passed in via the flow def init args fortheprocess. The flow does notmanage the lifecycle of thesechannels.
52
52
53
-
The step-fn may throw excepitonsfrom any arityand they will be handled by flow. Exceptions thrownfrom thetransition or transform arities,theexception will be logged on the flow's:error-chan.
53
+
`::flow/input-filter`, a predicate of cid, can be returned in the statefrom any arityto indicate a filter on the process input channel read set. For example, a step-fn that is waiting for a responsefrommultiple inputs might removethechannels that have already responded fromtheread-set until responses have been received from all.
54
54
55
55
###step-fn helpers
56
56
57
57
Some additional helpers exist to create step-fns from other forms:
58
58
59
-
*`lift*->step` - given a fn f taking one arg and returning a collection of non-nil values, creates a step-fn as needed by a process, with one input and one output (named:in and:out), and no state
59
+
*`lift*->step` - given a fn f taking one arg and returning a collection of non-nil values, creates a step-fn as needed by a process launcher, with one input and one output (named:in and:out), and no state
60
60
*`lift1->step` - like`lift*->step` but for functions that return a single value (when`nil`, yield no output)
61
61
*`map->step` - given a map with keys`:describe`,`:init`,`:transition`,`:transform` corresponding to the arities above, create a step-fn.
62
62
63
-
###Creating a process
63
+
###Creating a process launcher
64
64
65
-
Processes can be created using the[process](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-process) function, which takes a step-fn, and an option map with keys:
65
+
Process launchers can be created using the[process](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-process) function, which takes a step-fn, and an option map with keys:
66
66
67
67
*`::workload` - one of`:mixed`,`:io` or`:compute`
68
68
*`:compute-timeout-ms` - if:workload is:compute, this timeout (default 5000 msec) will be used when getting the return from the future - see below
69
69
70
-
A:workload supplied as an option to process will override any:workload returned by the:describe fn of the process. If neither are provded the default is:mixed.
70
+
A:workload supplied as an option to`process` will override any:workload returned by the:describe fn of the process launcher. If neither are provded the default is:mixed.
71
71
72
72
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_.
73
73
@@ -77,17 +77,19 @@ When :compute is specified, each call to transform will be run in a separate thr
77
77
78
78
When:compute is specified transform must not block!
79
79
80
+
Note that process launchers are defined by the[ProcLauncher](https://clojure.github.io/core.async/clojure.core.async.flow.spi.html#var-ProcLauncher) protocol. While you will typically use`process` to create a process launcher, advanced uses may also implement the protocol directly.
81
+
80
82
###Reloading
81
83
82
84
Because the step-fn is called in a loop, it is a good practice to define the step-fn in a var and use the var (`#'the-fn`) instead of the function value itself (`the-fn`). This practice supports interactive development by allowing the var to be rebound from the repl while the flow is running.
83
85
84
86
##Flow def
85
87
86
-
The step-fns are how you supply code for each process in the flow. The other thing you must supply is the flow configuration that ties together theprocs and the connections between them.
88
+
The step-fns are how you supply code for each process in the flow. The other thing you must supply is the flow configuration that ties together theproc launchers and the connections between them.
87
89
88
90
This flow definition is supplied to the[create-flow](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-create-flow) function and consists of a map with`:procs`,`:conns`, and optionally some workflow executors.
89
91
90
-
The`:procs` is a map of pid -> proc-def. The proc-def is a map with`:proc` (the processfunction), the`:args` (passed to the init arity of the step-fn), and the`:chan-opts` which can be used to specify channel properties.
92
+
The`:procs` is a map of pid -> proc-def. The proc-def is a map with`:proc` (the processlauncher), the`:args` (passed to the init arity of the step-fn), and the`:chan-opts` which can be used to specify channel properties.
91
93
92
94
The`:conns` is a collection of`[[from-pid outid] [to-pid inid]]` tuples. Inputs and outputs support multiple connections. When an output is connected multiple times, every connection will get every message, per`core.async/mult`.
93
95
@@ -116,7 +118,7 @@ When a flow is created, it starts in the resumed state. The following flow funct
116
118
* [pause-proc](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-pause-proc) - Pauses a single proc
117
119
* [resume-proc](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-resume-proc) - Resumes a single proc
118
120
119
-
You can alsouse these functions to ping the running processesare return their current state and status:
121
+
You can alsouse these functions to ping the running processesand return their current state and status:
120
122
121
123
* [ping](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-ping) - Pings all procs and returns a map of their status
122
124
* [ping-proc](https://clojure.github.io/core.async/clojure.core.async.flow.html#var-ping-proc) - Pings a single proce by pid and returns a map of status