README
¶
go-flow
A cancellable concurrent pattern for Go programming language
Go routine and channels facilitate developers to do concurrent programming. However, it is not easy for a beginner to write bug-free fan-out/fan-in go-routines. Especially when dealing with a complex flow net, make it cancellable is not that straightforward. Consider following situations:
There are 5 ways to exit from a go routine job:
- Successful return void or result(s)
- Expected error return
- Unexpected panic / error
- Job is timeout
- Job is cancelled from another go routine
There are 2 actions to deal with panic / error :
- quit the whole process whenever there is a panic
- only cancel the problematic go-routine branch (includes its sub-go-routines)
This package is aim to abstract the flow net, and give a panic-free solution.
GOAL
- User defines the flow net
- It completes the whole task or fail in all
- User should get notify whether there is a panic, or error, or job succeed
- User could define the timeout for the whole task
How-To
Note: this package requires go version 1.7+. Or you need import "golang.org/x/net/context" package manually if using a lower version of golang.
<1> define a flow net, together with a timeout duration
flow := NewFlowNet(1 * time.Millisecond)
<2> define a super start node (must-have), a super sink node (must-have), and several internal nodes (optional) Note that, all nodes must have a unique name tag. They are used by the flow control.
start := flow.InitStart("Start")
A := flow.InitNode("A")
B := flow.InitNode("B")
C := flow.InitSink("C")
<3> define actions for each node. The function signature is func() error. And each node could have multiple input and output channels, where all channels' signature is chan interface{}. User could identify input channel by using node.From(nodeName) function, and output channel by using node.To(nodeName)
start.Tk = func() error {
start.To("A") <- 1
start.To("B") <- "2"
return nil
}
A.Tk = func() error {
a := <-A.From("Start")
A.To("C") <- a
return nil
}
B.Tk = func() error {
bStr := <-B.From("Start")
switch bStr := bStr.(type) {
case string:
B.To("C") <- b
}
return nil
}
C.Tk = func() error {
a := <-C.From("A")
b := <-C.From("B")
// do something with a and b
C.ToSink() <- true // indicate job is done
return nil
}
<4> connect the dots
flow.Connect(start, A)
flow.Connect(start, B)
flow.Connect(A, C)
flow.Connect(B, C)
<5> run the flow
flow.Run()
<6> cleanup the flow after use it (optional)
flow.Cleanup()
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Background ¶
type Background struct {
Ctx context.Context // background context
Err chan<- error // write-only error channel
}
Flow's background, which defines its timed context and a global error channel.
type Flow ¶
type Flow struct {
Conn map[string]chan interface{} // edge name -> channel
Nodes map[string]*Node // node name -> node
Start *Node // super start node, not in Nodes map
TOut time.Duration // timeout definition
Done chan interface{} // done channel, named as "Done"
}
A flow is defined as: A map for all connections / edges; a map for all nodes; a super start node, which is regarded as flow's switch; a timeout duration, which defines max running time for the flow; and a Done channel, which indicate the flow is completed or not by using a bool value.
func (*Flow) Connect ¶
Connect two nodes in the flow. Channel will be named as {n1.Name}2{n2.Name} to indicate the direction.
func (*Flow) InitSink ¶
Init sink node. It will be added to flow.Nodes map. This is a special node, which has an output pointing to flow's Done channel (aka. super sink node).
type Node ¶
type Node struct {
Name string // node name, cannot be empty
Ps *Ports // node's interfaces
Tk func() error // node's job
Bg *Background // background, cannot be nil
}
Each node in the flow must have an unique name. It is used for labeled the connections / edges. The main task is defined in Tk function, which take Ps ports as input and output. In order to better control the task, user should pass in the flow's background.
func NewNode ¶
Return a new node with specified unique name, which should not be nil or empty string. It will help user to initialize ports too. Note that, 1) user should define its background and function task before using it; 2) "SUPER_SINK_NODE" is a special reserved name.
func (*Node) Run ¶
func (n *Node) Run()
Run the task defined in Node.Tk function. Normally, it should be called by flow.Run(), if this node is joined in the flow.
Source Files