flow

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.

---

Why go-flow?

Concurrent Go code often turns into boilerplate: goroutines, channels, waitgroups, select blocks, and error joins tangled with business logic.

go-flow lets you write workflows as explicit dependency graphs and linear pipelines:

• Readable & Maintainable: Business logic stays in normal Go functions. Timeouts, retries, fallbacks, and recovery are attached as step decorators.
• Graph-Based Construction: Declare DAG workflows with explicit edges (flow.Edge(fetchUser, processPayment)), catch cycles at startup, and export diagrams with plan.ToMermaid().
• Easy for Static Tools & AI: Because dependencies are declared explicitly in code instead of wired through runtime channel passing, tools and AI agents can read the full execution graph directly from the AST without grepping through goroutine closures.

---

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)

Runs steps in sequential order and stops on the first error encountered:

pipeline := flow.Seq(
    validateInput,
    fetchUserData,
    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	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
DAGWithReport	Executes DAG and collects per-node latency, status, and telemetry.	Guide · Example
Branch / When	Runs steps conditionally based on runtime context.	Guide · Example
Dynamic / .Dynamic()	Resolves and evaluates execution steps lazily at runtime.	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
Recovery	Middleware converting panics into standard Go errors.	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
.Wrap()	Attaches onion middleware (e.g. Recovery(), logging).	Guide · Example

---

Documentation & Examples

• GETTING_STARTED.md: Full guide, step-by-step tutorials, and patterns.
• examples/: 15+ runnable side-by-side examples comparing standard Go with go-flow.

---

License

MIT License

Documentation

Overview

Package flow provides structured concurrency and workflow pipelines for flow.

Package flow provides structured concurrency and workflow pipelines for flow.

Package flow provides structured concurrency and workflow pipelines for flow.

Package flow provides structured concurrency and workflow pipelines for flow.

Package flow provides structured concurrency and workflow pipelines for flow.

Index

• Variables
• func DAGEdgesNWithReport[T context.Context](limit int, connections ...DAGConnection[T]) func(ctx T) (*DAGReport, error)
• func DAGEdgesToMermaid[T context.Context](connections ...DAGConnection[T]) (string, error)
• func DAGEdgesWithReport[T context.Context](connections ...DAGConnection[T]) func(ctx T) (*DAGReport, error)
• func DAGNWithReport[T context.Context](limit int, nodes ...*DAGNode[T]) func(ctx T) (*DAGReport, error)
• func DAGToMermaid[T context.Context](nodes ...*DAGNode[T]) (string, error)
• func DAGWithReport[T context.Context](nodes ...*DAGNode[T]) func(ctx T) (*DAGReport, 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
  • func Edge[T context.Context](from Step[T], to ...Step[T]) DAGConnection[T]
• type DAGEdgeBuilder
  • func From[T context.Context](from Step[T]) DAGEdgeBuilder[T]
  • func (b DAGEdgeBuilder[T]) To(to ...Step[T]) DAGConnection[T]
• type DAGNode
  • func Node[T context.Context](name string, step Step[T]) *DAGNode[T]
  • func (n *DAGNode[T]) After(deps ...string) *DAGNode[T]
  • func (n *DAGNode[T]) Unless(predicate func(ctx T) bool) *DAGNode[T]
  • func (n *DAGNode[T]) When(predicate func(ctx T) bool) *DAGNode[T]
  • func (n *DAGNode[T]) WithFallback(fallback Step[T]) *DAGNode[T]
  • func (n *DAGNode[T]) WithRetry(attempts int, delay ...time.Duration) *DAGNode[T]
  • func (n *DAGNode[T]) WithTimeout(d time.Duration) *DAGNode[T]
• type DAGPlan
  • func NewDAG[T context.Context](nodes ...*DAGNode[T]) *DAGPlan[T]
  • func NewDAGEdges[T context.Context](connections ...DAGConnection[T]) (*DAGPlan[T], error)
  • func (p *DAGPlan[T]) ExecNWithReport(limit int, ctx T) (*DAGReport, error)
  • func (p *DAGPlan[T]) ExecWithReport(ctx T) (*DAGReport, error)
  • func (p *DAGPlan[T]) Step() Step[T]
  • func (p *DAGPlan[T]) StepN(limit int) Step[T]
  • func (p *DAGPlan[T]) StepNWithReport(limit int) func(ctx T) (*DAGReport, error)
  • func (p *DAGPlan[T]) StepWithReport() func(ctx T) (*DAGReport, error)
  • func (p *DAGPlan[T]) ToMermaid() (string, error)
  • func (p *DAGPlan[T]) Validate() error
• type DAGReport
  • func (r *DAGReport) Failed() []NodeReport
  • func (r *DAGReport) Node(name string) *NodeReport
  • func (r *DAGReport) Skipped() []NodeReport
  • func (r *DAGReport) Successful() []NodeReport
• type Middleware
  • func Recovery[T context.Context]() Middleware[T]
• type NodeReport
• type NodeStatus
• 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 Dynamic[T context.Context](fn func(ctx T) Step[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]) Dynamic(fn func(ctx T) Step[T]) 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]) Retry(attempts int, delay time.Duration) 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]

Examples

• Branch
• Chunk
• DAG
• DAG (Conditional)
• DAGEdges
• DAGN
• DAGToMermaid
• DAGWithReport
• Each
• Go
• GoN
• NewDAG
• Once
• Pipe
• PipeSeq
• Race
• Recovery
• Seq
• Step.Fallback
• Step.Retry
• Step.Timeout
• Step.When
• Step.Wrap

Variables

var (
	// ErrDAGCycle defines ErrDAGCycle.
	ErrDAGCycle = errors.New("cycle detected in DAG")
	// ErrDAGUnknownDependency defines ErrDAGUnknownDependency.
	ErrDAGUnknownDependency = errors.New("unknown dependency in DAG")
	// ErrDAGDuplicateNode defines ErrDAGDuplicateNode.
	ErrDAGDuplicateNode = errors.New("duplicate node name in DAG")
	// ErrDAGNilNode defines ErrDAGNilNode.
	ErrDAGNilNode = errors.New("nil node in DAG")
	// ErrDAGEmptyNodeName defines ErrDAGEmptyNodeName.
	ErrDAGEmptyNodeName = errors.New("empty node name in DAG")
)

Functions

func DAGEdgesNWithReport

func DAGEdgesNWithReport[T context.Context](limit int, connections ...DAGConnection[T]) func(ctx T) (*DAGReport, error)

DAGEdgesNWithReport performs the DAGEdgesNWithReport operation.

func DAGEdgesToMermaid

func DAGEdgesToMermaid[T context.Context](connections ...DAGConnection[T]) (string, error)

DAGEdgesToMermaid performs the DAGEdgesToMermaid operation.

func DAGEdgesWithReport

func DAGEdgesWithReport[T context.Context](connections ...DAGConnection[T]) func(ctx T) (*DAGReport, error)

DAGEdgesWithReport performs the DAGEdgesWithReport operation.

func DAGNWithReport

func DAGNWithReport[T context.Context](limit int, nodes ...*DAGNode[T]) func(ctx T) (*DAGReport, error)

DAGNWithReport performs the DAGNWithReport operation.

func DAGToMermaid

func DAGToMermaid[T context.Context](nodes ...*DAGNode[T]) (string, error)

DAGToMermaid performs the DAGToMermaid operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	user := flow.Node("user", func(ctx context.Context) error { return nil })
	payment := flow.Node("payment", func(ctx context.Context) error { return nil }).After("user")

	mermaid, err := flow.DAGToMermaid(user, payment)
	if err != nil {
		panic(err)
	}

	fmt.Println(len(mermaid) > 0)
}

func DAGWithReport

func DAGWithReport[T context.Context](nodes ...*DAGNode[T]) func(ctx T) (*DAGReport, error)

DAGWithReport performs the DAGWithReport operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	user := flow.Node("user", func(ctx context.Context) error { return nil })
	payment := flow.Node("payment", func(ctx context.Context) error { return nil }).After("user")

	exec := flow.DAGWithReport(user, payment)
	report, err := exec(context.Background())
	if err != nil {
		panic(err)
	}

	for _, n := range report.Nodes {
		fmt.Printf("%s: %s\n", n.Name, n.Status)
	}
}

func Pipe2

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)

Pipe2 performs the Pipe2 operation.

func Pipe3

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),
	f3 func(ctx T, c C) (D, error),
) func(ctx T, a A) (D, error)

Pipe3 performs the Pipe3 operation.

Types

type DAGConnection

type DAGConnection[T context.Context] struct {
	// contains filtered or unexported fields
}

DAGConnection represents DAGConnection.

func Edge

func Edge[T context.Context](from Step[T], to ...Step[T]) DAGConnection[T]

Edge performs the Edge operation.

type DAGEdgeBuilder

type DAGEdgeBuilder[T context.Context] struct {
	// contains filtered or unexported fields
}

DAGEdgeBuilder represents DAGEdgeBuilder.

func From

func From[T context.Context](from Step[T]) DAGEdgeBuilder[T]

From performs the From operation.

func (DAGEdgeBuilder[T]) To

func (b DAGEdgeBuilder[T]) To(to ...Step[T]) DAGConnection[T]

To executes the To operation.

type DAGNode

type DAGNode[T context.Context] struct {
	// contains filtered or unexported fields
}

DAGNode represents DAGNode.

func Node

func Node[T context.Context](name string, step Step[T]) *DAGNode[T]

Node performs the Node operation.

func (*DAGNode[T]) After

func (n *DAGNode[T]) After(deps ...string) *DAGNode[T]

After executes the After operation.

func (*DAGNode[T]) Unless

func (n *DAGNode[T]) Unless(predicate func(ctx T) bool) *DAGNode[T]

Unless executes the Unless operation.

func (*DAGNode[T]) When

func (n *DAGNode[T]) When(predicate func(ctx T) bool) *DAGNode[T]

When executes the When operation.

func (*DAGNode[T]) WithFallback

func (n *DAGNode[T]) WithFallback(fallback Step[T]) *DAGNode[T]

WithFallback executes the WithFallback operation.

func (*DAGNode[T]) WithRetry

func (n *DAGNode[T]) WithRetry(attempts int, delay ...time.Duration) *DAGNode[T]

WithRetry executes the WithRetry operation.

func (*DAGNode[T]) WithTimeout

func (n *DAGNode[T]) WithTimeout(d time.Duration) *DAGNode[T]

WithTimeout executes the WithTimeout operation.

type DAGPlan

type DAGPlan[T context.Context] struct {
	// contains filtered or unexported fields
}

DAGPlan represents DAGPlan.

func NewDAG

func NewDAG[T context.Context](nodes ...*DAGNode[T]) *DAGPlan[T]

NewDAG performs the NewDAG operation.

Example

package main

import (
	"context"

	"github.com/ntkwan/go-flow"
)

func main() {
	user := flow.Node("user", func(ctx context.Context) error { return nil })
	payment := flow.Node("payment", func(ctx context.Context) error { return nil }).After("user")

	plan := flow.NewDAG(user, payment)
	if err := plan.Validate(); err != nil {
		panic(err)
	}

	step := plan.Step()
	_ = step(context.Background())
}

func NewDAGEdges

func NewDAGEdges[T context.Context](connections ...DAGConnection[T]) (*DAGPlan[T], error)

NewDAGEdges performs the NewDAGEdges operation.

func (*DAGPlan[T]) ExecNWithReport

func (p *DAGPlan[T]) ExecNWithReport(limit int, ctx T) (*DAGReport, error)

ExecNWithReport executes the ExecNWithReport operation.

func (*DAGPlan[T]) ExecWithReport

func (p *DAGPlan[T]) ExecWithReport(ctx T) (*DAGReport, error)

ExecWithReport executes the ExecWithReport operation.

func (*DAGPlan[T]) Step

func (p *DAGPlan[T]) Step() Step[T]

Step executes the Step operation.

func (*DAGPlan[T]) StepN

func (p *DAGPlan[T]) StepN(limit int) Step[T]

StepN executes the StepN operation.

func (*DAGPlan[T]) StepNWithReport

func (p *DAGPlan[T]) StepNWithReport(limit int) func(ctx T) (*DAGReport, error)

StepNWithReport executes the StepNWithReport operation.

func (*DAGPlan[T]) StepWithReport

func (p *DAGPlan[T]) StepWithReport() func(ctx T) (*DAGReport, error)

StepWithReport executes the StepWithReport operation.

func (*DAGPlan[T]) ToMermaid

func (p *DAGPlan[T]) ToMermaid() (string, error)

ToMermaid executes the ToMermaid operation.

func (*DAGPlan[T]) Validate

func (p *DAGPlan[T]) Validate() error

Validate executes the Validate operation.

type DAGReport

type DAGReport struct {
	Duration  time.Duration
	StartTime time.Time
	EndTime   time.Time
	Nodes     []NodeReport
	// contains filtered or unexported fields
}

DAGReport represents DAGReport.

func (*DAGReport) Failed

func (r *DAGReport) Failed() []NodeReport

Failed executes the Failed operation.

func (*DAGReport) Node

func (r *DAGReport) Node(name string) *NodeReport

Node executes the Node operation.

func (*DAGReport) Skipped

func (r *DAGReport) Skipped() []NodeReport

Skipped executes the Skipped operation.

func (*DAGReport) Successful

func (r *DAGReport) Successful() []NodeReport

Successful executes the Successful operation.

type Middleware

type Middleware[T context.Context] func(next Step[T]) Step[T]

Middleware represents Middleware.

func Recovery

func Recovery[T context.Context]() Middleware[T]

Recovery performs the Recovery operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	riskyStep := flow.Step[context.Context](func(ctx context.Context) error {
		panic("database connection lost")
	})

	guarded := riskyStep.Wrap(flow.Recovery[context.Context]())
	err := guarded.Exec(context.Background())
	fmt.Println("Recovered from panic:", err != nil)
}

type NodeReport

type NodeReport struct {
	Name      string
	Status    NodeStatus
	Duration  time.Duration
	StartTime time.Time
	EndTime   time.Time
	Err       error
}

NodeReport represents NodeReport.

type NodeStatus

type NodeStatus string

NodeStatus represents NodeStatus.

const (
	// NodeStatusSuccess defines NodeStatusSuccess.
	NodeStatusSuccess NodeStatus = "SUCCESS"
	// NodeStatusFailed defines NodeStatusFailed.
	NodeStatusFailed NodeStatus = "FAILED"
	// NodeStatusSkipped defines NodeStatusSkipped.
	NodeStatusSkipped NodeStatus = "SKIPPED"
)

type Step

type Step[T context.Context] func(ctx T) error

Step represents Step.

func Branch

func Branch[T context.Context](condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]

Branch performs the Branch operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	type RequestContext struct {
		context.Context
		IsAdmin bool
	}

	adminStep := func(ctx *RequestContext) error {
		fmt.Println("Admin flow")
		return nil
	}
	userStep := func(ctx *RequestContext) error {
		fmt.Println("User flow")
		return nil
	}

	route := flow.Branch(
		func(ctx *RequestContext) bool { return ctx.IsAdmin },
		adminStep,
		userStep,
	)

	ctx := &RequestContext{Context: context.Background(), IsAdmin: true}
	_ = route(ctx)
}

func Chunk

func Chunk[T context.Context, V any](seq iter.Seq[V], size int, step func(ctx T, batch []V) error) Step[T]

Chunk performs the Chunk operation.

Example

package main

import (
	"context"
	"fmt"
	"slices"

	"github.com/ntkwan/go-flow"
)

func main() {
	numbers := []int{1, 2, 3, 4, 5, 6, 7}
	batchProcess := flow.Chunk(slices.Values(numbers), 3, func(ctx context.Context, batch []int) error {
		fmt.Printf("Batch of %d items: %v\n", len(batch), batch)
		return nil
	})

	_ = batchProcess(context.Background())
}

func Chunk2

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]

Chunk2 performs the Chunk2 operation.

func DAG

func DAG[T context.Context](nodes ...*DAGNode[T]) Step[T]

DAG performs the DAG operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	user := flow.Node("user", func(ctx context.Context) error {
		fmt.Println("User loaded")
		return nil
	})
	inventory := flow.Node("inventory", func(ctx context.Context) error {
		fmt.Println("Inventory reserved")
		return nil
	})
	payment := flow.Node("payment", func(ctx context.Context) error {
		fmt.Println("Payment charged")
		return nil
	}).After("user", "inventory")

	graph := flow.DAG(user, inventory, payment)
	_ = graph(context.Background())
}

Example (Conditional)

package main

import (
	"context"

	"github.com/ntkwan/go-flow"
)

func main() {
	type userContext struct {
		context.Context
		isPremium bool
	}

	baseTask := flow.Node("base", func(ctx *userContext) error { return nil })
	bonusTask := flow.Node("bonus", func(ctx *userContext) error { return nil }).
		After("base").
		When(func(ctx *userContext) bool { return ctx.isPremium })

	graph := flow.DAG(baseTask, bonusTask)
	_ = graph(&userContext{Context: context.Background(), isPremium: false})
}

func DAGEdges

func DAGEdges[T context.Context](connections ...DAGConnection[T]) Step[T]

DAGEdges performs the DAGEdges operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	fetchUser := func(ctx context.Context) error {
		fmt.Println("Fetch user")
		return nil
	}
	chargePayment := func(ctx context.Context) error {
		fmt.Println("Charge payment")
		return nil
	}

	graph := flow.DAGEdges(
		flow.From(fetchUser).To(chargePayment),
	)

	_ = graph(context.Background())
}

func DAGEdgesN

func DAGEdgesN[T context.Context](limit int, connections ...DAGConnection[T]) Step[T]

DAGEdgesN performs the DAGEdgesN operation.

func DAGN

func DAGN[T context.Context](limit int, nodes ...*DAGNode[T]) Step[T]

DAGN performs the DAGN operation.

Example

package main

import (
	"context"

	"github.com/ntkwan/go-flow"
)

func main() {
	n1 := flow.Node("task1", func(ctx context.Context) error { return nil })
	n2 := flow.Node("task2", func(ctx context.Context) error { return nil })
	n3 := flow.Node("task3", func(ctx context.Context) error { return nil })

	boundedGraph := flow.DAGN(2, n1, n2, n3)
	_ = boundedGraph(context.Background())
}

func Dynamic

func Dynamic[T context.Context](fn func(ctx T) Step[T]) Step[T]

Dynamic performs the Dynamic operation.

func Each

func Each[T context.Context, V any](seq iter.Seq[V], step func(ctx T, item V) error) Step[T]

Each performs the Each operation.

Example

package main

import (
	"context"
	"fmt"
	"slices"

	"github.com/ntkwan/go-flow"
)

func main() {
	items := slices.Values([]string{"apple", "banana", "cherry"})
	process := flow.Each(items, func(ctx context.Context, item string) error {
		fmt.Println("Processing:", item)
		return nil
	})

	_ = process(context.Background())
}

func Each2

func Each2[T context.Context, K, V any](seq iter.Seq2[K, V], step func(ctx T, key K, val V) error) Step[T]

Each2 performs the Each2 operation.

func Go

func Go[T context.Context](steps ...Step[T]) Step[T]

Go performs the Go operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	stepA := func(ctx context.Context) error {
		fmt.Println("Task A")
		return nil
	}
	stepB := func(ctx context.Context) error {
		fmt.Println("Task B")
		return nil
	}

	parallel := flow.Go(stepA, stepB)
	_ = parallel(context.Background())
}

func GoN

func GoN[T context.Context](limit int, steps ...Step[T]) Step[T]

GoN performs the GoN operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	tasks := make([]flow.Step[context.Context], 5)
	for i := range tasks {
		idx := i
		tasks[i] = func(ctx context.Context) error {
			fmt.Printf("Worker task %d\n", idx)
			return nil
		}
	}

	bounded := flow.GoN(2, tasks...)
	_ = bounded(context.Background())
}

func Once

func Once[T context.Context](step Step[T]) Step[T]

Once performs the Once operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	initDB := flow.Once(func(ctx context.Context) error {
		fmt.Println("Database initialized once")
		return nil
	})

	step1 := flow.Seq(initDB, func(ctx context.Context) error { return nil })
	step2 := flow.Seq(initDB, func(ctx context.Context) error { return nil })

	pipeline := flow.Go(step1, step2)
	_ = pipeline(context.Background())
}

func Pipe

func Pipe[T context.Context, In, Out any](
	fn func(ctx T, in In) (Out, error),
	get func(ctx T) In,
	set func(ctx T, out Out),
) Step[T]

Pipe performs the Pipe operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	type State struct {
		Count int
	}
	st := &State{Count: 5}

	doubleCount := flow.Pipe(
		func(ctx context.Context, in int) (int, error) {
			return in * 2, nil
		},
		func(ctx context.Context) int {
			return st.Count
		},
		func(ctx context.Context, out int) {
			st.Count = out
		},
	)

	_ = doubleCount(context.Background())
	fmt.Printf("Result: %d\n", st.Count)
}

func PipeSeq

func PipeSeq[T context.Context, In, Out any](
	seq iter.Seq[In],
	transform func(ctx T, item In) (Out, error),
	sink func(ctx T, item Out) error,
) Step[T]

PipeSeq performs the PipeSeq operation.

Example

package main

import (
	"context"
	"fmt"
	"slices"

	"github.com/ntkwan/go-flow"
)

func main() {
	numbers := slices.Values([]int{1, 2, 3})
	stream := flow.PipeSeq(
		numbers,
		func(ctx context.Context, in int) (string, error) {
			return fmt.Sprintf("val-%d", in*10), nil
		},
		func(ctx context.Context, out string) error {
			fmt.Println(out)
			return nil
		},
	)

	_ = stream(context.Background())
}

func PipeSeq2

func PipeSeq2[T context.Context, K, V, Out any](
	seq iter.Seq2[K, V],
	transform func(ctx T, key K, val V) (Out, error),
	sink func(ctx T, item Out) error,
) Step[T]

PipeSeq2 performs the PipeSeq2 operation.

func Race

func Race[T context.Context](steps ...Step[T]) Step[T]

Race performs the Race operation.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ntkwan/go-flow"
)

func main() {
	server1 := func(ctx context.Context) error {
		time.Sleep(10 * time.Millisecond)
		fmt.Println("Server 1 responded")
		return nil
	}
	server2 := func(ctx context.Context) error {
		fmt.Println("Server 2 responded fastest")
		return nil
	}

	fastest := flow.Race(server1, server2)
	_ = fastest(context.Background())
}

func Seq

func Seq[T context.Context](steps ...Step[T]) Step[T]

Seq performs the Seq operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	step1 := func(ctx context.Context) error {
		fmt.Println("Step 1")
		return nil
	}
	step2 := func(ctx context.Context) error {
		fmt.Println("Step 2")
		return nil
	}

	pipeline := flow.Seq(step1, step2)
	_ = pipeline(context.Background())
}

func (Step[T]) Branch

func (s Step[T]) Branch(condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]

Branch executes the Branch operation.

func (Step[T]) Dynamic

func (s Step[T]) Dynamic(fn func(ctx T) Step[T]) Step[T]

Dynamic executes the Dynamic operation.

func (Step[T]) Exec

func (s Step[T]) Exec(ctx T) error

Exec executes the Exec operation.

func (Step[T]) Fallback

func (s Step[T]) Fallback(fallback Step[T]) Step[T]

Fallback executes the Fallback operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	failingPrimary := flow.Step[context.Context](func(ctx context.Context) error {
		return fmt.Errorf("primary service down")
	})
	backupService := flow.Step[context.Context](func(ctx context.Context) error {
		fmt.Println("Backup service handled request")
		return nil
	})

	protected := failingPrimary.Fallback(backupService)
	_ = protected.Exec(context.Background())
}

func (Step[T]) Go

func (s Step[T]) Go(steps ...Step[T]) Step[T]

Go executes the Go operation.

func (Step[T]) GoN

func (s Step[T]) GoN(limit int, steps ...Step[T]) Step[T]

GoN executes the GoN operation.

func (Step[T]) Once

func (s Step[T]) Once() Step[T]

Once executes the Once operation.

func (Step[T]) Race

func (s Step[T]) Race(steps ...Step[T]) Step[T]

Race executes the Race operation.

func (Step[T]) Retry

func (s Step[T]) Retry(attempts int, delay time.Duration) Step[T]

Retry executes the Retry operation.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ntkwan/go-flow"
)

func main() {
	attempts := 0
	flaky := flow.Step[context.Context](func(ctx context.Context) error {
		attempts++
		if attempts < 3 {
			return fmt.Errorf("temporary failure %d", attempts)
		}
		fmt.Println("Succeeded on attempt 3")
		return nil
	})

	resilient := flaky.Retry(3, 10*time.Millisecond)
	_ = resilient.Exec(context.Background())
}

func (Step[T]) Timeout

func (s Step[T]) Timeout(d time.Duration) Step[T]

Timeout executes the Timeout operation.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ntkwan/go-flow"
)

func main() {
	slow := flow.Step[context.Context](func(ctx context.Context) error {
		time.Sleep(50 * time.Millisecond)
		return nil
	})

	timed := slow.Timeout(10 * time.Millisecond)
	err := timed.Exec(context.Background())
	fmt.Println("Error:", err != nil)
}

func (Step[T]) Unless

func (s Step[T]) Unless(predicate func(ctx T) bool) Step[T]

Unless executes the Unless operation.

func (Step[T]) When

func (s Step[T]) When(predicate func(ctx T) bool) Step[T]

When executes the When operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	type OrderContext struct {
		context.Context
		IsVIP bool
	}

	vipGift := flow.Step[*OrderContext](func(ctx *OrderContext) error {
		fmt.Println("VIP gift applied")
		return nil
	}).When(func(ctx *OrderContext) bool {
		return ctx.IsVIP
	})

	ctx := &OrderContext{Context: context.Background(), IsVIP: true}
	_ = vipGift.Exec(ctx)
}

func (Step[T]) Wrap

func (s Step[T]) Wrap(middlewares ...Middleware[T]) Step[T]

Wrap executes the Wrap operation.

Example

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	base := flow.Step[context.Context](func(ctx context.Context) error {
		fmt.Println("Core logic")
		return nil
	})

	loggingMiddleware := func(next flow.Step[context.Context]) flow.Step[context.Context] {
		return func(ctx context.Context) error {
			fmt.Println("Before step")
			err := next(ctx)
			fmt.Println("After step")
			return err
		}
	}

	wrapped := base.Wrap(loggingMiddleware)
	_ = wrapped.Exec(context.Background())
}