README
¶
go-flow
A structured concurrency and workflow library for Go (1.24+). Built with generics, standard iterators, and zero external dependencies.
Inspired by C++26 std::execution (NVIDIA/stdexec), go-flow replaces manual goroutine management and channel synchronization with declarative, cancellable execution pipelines and dependency graphs.
Detailed guides and patterns are available in GETTING_STARTED.md.
Installation
go get github.com/ntkwan/go-flow
Mental Model
graph LR
subgraph Sequential [1. Linear Seq]
A1[Step 1] --> A2[Step 2] --> A3[Step 3]
end
subgraph Parallel [2. Concurrent Go / GoN]
B1[Worker 1]
B2[Worker 2]
B3[Worker 3]
end
subgraph Speculative [3. Speculative Race]
R1[Server 1]
R2[Server 2]
R3[Server 3]
end
subgraph DAG [4. Dependency DAG]
D1[Fetch User] --> D3[Process Payment]
D2[Fetch Cart] --> D3
D3 --> D4[Send Receipt]
end
Basic Usage
Sequential Execution (Seq / .Then())
Runs steps in sequential order and stops on the first error encountered:
pipeline := flow.Seq(
validateInput,
fetchUserData,
saveRecord,
)
// Or using method chaining:
pipeline = flow.Step[context.Context](validateInput).
Then(fetchUserData).
Then(saveRecord)
if err := pipeline.Exec(ctx); err != nil {
log.Println("Pipeline failed:", err)
}
Concurrent Execution (Go / .Go())
Runs steps concurrently and joins all encountered errors:
fetchUserProfile := flow.Go(
fetchAccount,
fetchOrders,
fetchPreferences,
)
if err := fetchUserProfile.Exec(ctx); err != nil {
log.Println("Concurrent execution failed:", err)
}
Bounded Concurrency (GoN / .GoN())
Runs steps concurrently with a worker limit to control resource utilization:
processBatch := flow.GoN(4, workerTasks...)
if err := processBatch.Exec(ctx); err != nil {
log.Println("Batch processing failed:", err)
}
Dependency Graph (DAG)
validateNode := flow.Node("validate-order", validateOrder)
userNode := flow.Node("fetch-user", fetchUser).After("validate-order")
inventoryNode := flow.Node("fetch-inventory", fetchInventory).After("validate-order")
discountNode := flow.Node("calculate-discounts", calculateDiscounts).After("fetch-user")
paymentNode := flow.Node("process-payment", processPayment).After("calculate-discounts", "fetch-inventory")
updateInvNode := flow.Node("update-inventory", updateInventory).After("process-payment")
invoiceNode := flow.Node("generate-invoice", generateInvoice).After("process-payment")
auditNode := flow.Node("audit-log", auditLog).After("process-payment")
notifyNode := flow.Node("notify-customer", notifyCustomer).After("generate-invoice")
dispatchNode := flow.Node("dispatch-warehouse", dispatchWarehouse).After("update-inventory")
plan := flow.NewDAG(
validateNode, userNode, inventoryNode, discountNode, paymentNode,
updateInvNode, invoiceNode, auditNode, notifyNode, dispatchNode,
)
// Validate graph topology before execution
if err := plan.Validate(); err != nil {
log.Fatal(err)
}
// Execute as Step[T]
workflow := plan.Step()
if err := workflow(ctx); err != nil {
log.Println("Workflow failed:", err)
}
graph TD
validate_order["validate-order"] --> fetch_user["fetch-user"]
validate_order["validate-order"] --> fetch_inventory["fetch-inventory"]
fetch_user["fetch-user"] --> calculate_discounts["calculate-discounts"]
calculate_discounts["calculate-discounts"] --> process_payment["process-payment"]
fetch_inventory["fetch-inventory"] --> process_payment["process-payment"]
process_payment["process-payment"] --> update_inventory["update-inventory"]
process_payment["process-payment"] --> generate_invoice["generate-invoice"]
process_payment["process-payment"] --> audit_log["audit-log"]
generate_invoice["generate-invoice"] --> notify_customer["notify-customer"]
update_inventory["update-inventory"] --> dispatch_warehouse["dispatch-warehouse"]
Core Functions
| Function / Method | Description | Guide & Examples |
|---|---|---|
Seq / .Then() |
Runs steps in sequential order; stops on first error. | Guide · Example |
Go / .Go() |
Runs steps concurrently; joins all errors. | Guide · Example |
GoN / .GoN() |
Runs steps concurrently with a worker limit. | Guide · Example |
Race / .Race() |
Runs steps concurrently; returns on first success and cancels losers. | Guide · Example |
DAG / DAGN |
Runs steps according to a directed dependency graph. | Guide · Example |
Branch / When |
Runs steps conditionally based on runtime context. | Guide · Example |
Pipe / PipeSeq |
Transforms typed values across steps and iterators. | Guide · Example |
Each / Chunk |
Iterates over Go 1.23+ iter.Seq sequences or batches slices. |
Guide · Example |
Once / .Once() |
Ensures a step executes at most once across parallel branches. | Guide · Example |
.Retry() |
Retries a failing step up to $N$ attempts with backoff delay. | Guide · Example |
.Timeout() |
Applies a timeout deadline to step execution. | Guide · Example |
.Fallback() |
Runs an alternate step if the primary step fails. | Guide · Example |
.Catch() |
Handles or transforms step errors. | Guide · Example |
.Recover() |
Catches panics and converts them into standard Go errors. | Guide · Example |
Documentation & Examples
- GETTING_STARTED.md: Full guide, step-by-step tutorials, and patterns.
examples/: 14+ runnable side-by-side examples comparing standard Go withgo-flow.
License
MIT License
Documentation
¶
Index ¶
- Variables
- func DAGEdgesToDOT[T context.Context](connections ...DAGConnection[T]) (string, error)
- func DAGEdgesToMermaid[T context.Context](connections ...DAGConnection[T]) (string, error)
- func DAGToDOT[T context.Context](nodes ...*DAGNode[T]) (string, error)
- func DAGToMermaid[T context.Context](nodes ...*DAGNode[T]) (string, error)
- func Pipe2[T context.Context, A, B, C any](f1 func(ctx T, a A) (B, error), f2 func(ctx T, b B) (C, error)) func(ctx T, a A) (C, error)
- func Pipe3[T context.Context, A, B, C, D any](f1 func(ctx T, a A) (B, error), f2 func(ctx T, b B) (C, error), ...) func(ctx T, a A) (D, error)
- type DAGConnection
- type DAGEdgeBuilder
- type DAGNode
- type DAGPlan
- type Middleware
- type Step
- func Branch[T context.Context](condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]
- func Chunk[T context.Context, V any](seq iter.Seq[V], size int, step func(ctx T, batch []V) error) Step[T]
- func Chunk2[T context.Context, K, V any](seq iter.Seq2[K, V], size int, step func(ctx T, keys []K, vals []V) error) Step[T]
- func DAG[T context.Context](nodes ...*DAGNode[T]) Step[T]
- func DAGEdges[T context.Context](connections ...DAGConnection[T]) Step[T]
- func DAGEdgesN[T context.Context](limit int, connections ...DAGConnection[T]) Step[T]
- func DAGN[T context.Context](limit int, nodes ...*DAGNode[T]) Step[T]
- func Each[T context.Context, V any](seq iter.Seq[V], step func(ctx T, item V) error) Step[T]
- func Each2[T context.Context, K, V any](seq iter.Seq2[K, V], step func(ctx T, key K, val V) error) Step[T]
- func Go[T context.Context](steps ...Step[T]) Step[T]
- func GoN[T context.Context](limit int, steps ...Step[T]) Step[T]
- func Once[T context.Context](step Step[T]) Step[T]
- func Pipe[T context.Context, In, Out any](fn func(ctx T, in In) (Out, error), get func(ctx T) In, ...) Step[T]
- func PipeSeq[T context.Context, In, Out any](seq iter.Seq[In], transform func(ctx T, item In) (Out, error), ...) Step[T]
- func PipeSeq2[T context.Context, K, V, Out any](seq iter.Seq2[K, V], transform func(ctx T, key K, val V) (Out, error), ...) Step[T]
- func Race[T context.Context](steps ...Step[T]) Step[T]
- func Seq[T context.Context](steps ...Step[T]) Step[T]
- func (s Step[T]) Branch(condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]
- func (s Step[T]) Catch(handler func(ctx T, err error) error) Step[T]
- func (s Step[T]) Exec(ctx T) error
- func (s Step[T]) Fallback(fallback Step[T]) Step[T]
- func (s Step[T]) Go(steps ...Step[T]) Step[T]
- func (s Step[T]) GoN(limit int, steps ...Step[T]) Step[T]
- func (s Step[T]) Once() Step[T]
- func (s Step[T]) Race(steps ...Step[T]) Step[T]
- func (s Step[T]) Recover() Step[T]
- func (s Step[T]) Retry(attempts int, delay time.Duration) Step[T]
- func (s Step[T]) Then(steps ...Step[T]) Step[T]
- func (s Step[T]) Timeout(d time.Duration) Step[T]
- func (s Step[T]) Unless(predicate func(ctx T) bool) Step[T]
- func (s Step[T]) When(predicate func(ctx T) bool) Step[T]
- func (s Step[T]) Wrap(middlewares ...Middleware[T]) Step[T]
Constants ¶
This section is empty.
Variables ¶
View Source
var ( ErrDAGCycle = errors.New("cycle detected in DAG") ErrDAGUnknownDependency = errors.New("unknown dependency in DAG") ErrDAGDuplicateNode = errors.New("duplicate node name in DAG") ErrDAGNilNode = errors.New("nil node in DAG") ErrDAGEmptyNodeName = errors.New("empty node name in DAG") )
Functions ¶
func DAGEdgesToDOT ¶ added in v0.4.0
func DAGEdgesToDOT[T context.Context](connections ...DAGConnection[T]) (string, error)
func DAGEdgesToMermaid ¶ added in v0.4.0
func DAGEdgesToMermaid[T context.Context](connections ...DAGConnection[T]) (string, error)
func DAGToMermaid ¶ added in v0.4.0
Types ¶
type DAGConnection ¶
type DAGEdgeBuilder ¶
func (DAGEdgeBuilder[T]) To ¶
func (b DAGEdgeBuilder[T]) To(to ...Step[T]) DAGConnection[T]
type DAGPlan ¶ added in v0.4.0
func NewDAGEdges ¶ added in v0.4.0
func NewDAGEdges[T context.Context](connections ...DAGConnection[T]) (*DAGPlan[T], error)
type Step ¶
func DAGEdgesN ¶
func DAGEdgesN[T context.Context](limit int, connections ...DAGConnection[T]) Step[T]
func (Step[T]) Wrap ¶
func (s Step[T]) Wrap(middlewares ...Middleware[T]) Step[T]
Source Files
Click to show internal directories.
Click to hide internal directories.