Help us shape the future of Go! Take the 2026 Go Developer Survey

flow

package module
v0.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 5, 2026 License: MIT Imports: 8 Imported by: 0

README

go-flow

CI Go Reference Go Version Release Coverage

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
DAGWithReport Executes DAG and collects per-node latency, status, and telemetry. 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 with go-flow.

License

MIT License

Documentation

Index

Examples

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 DAGEdgesNWithReport added in v0.6.0 

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

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 DAGEdgesWithReport added in v0.6.0 

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

func DAGNWithReport added in v0.6.0 

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

func DAGToDOT added in v0.4.0 

func DAGToDOT[T context.Context](nodes ...*DAGNode[T]) (string, error)
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")

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

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

func DAGToMermaid added in v0.4.0 

func DAGToMermaid[T context.Context](nodes ...*DAGNode[T]) (string, error)
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 added in v0.6.0 

func DAGWithReport[T context.Context](nodes ...*DAGNode[T]) func(ctx T) (*DAGReport, error)
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 added in v0.3.0 

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 added in v0.3.0 

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)

Types

type DAGConnection 

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

func Edge 

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

type DAGEdgeBuilder 

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

func From 

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

func (DAGEdgeBuilder[T]) To 

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

type DAGNode 

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

func Node 

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

func (*DAGNode[T]) After 

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

func (*DAGNode[T]) Unless added in v0.6.0 

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

func (*DAGNode[T]) When added in v0.6.0 

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

func (*DAGNode[T]) WithCatch added in v0.6.0 

func (n *DAGNode[T]) WithCatch(handler func(ctx T, err error) error) *DAGNode[T]

func (*DAGNode[T]) WithFallback added in v0.6.0 

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

func (*DAGNode[T]) WithRecover added in v0.6.0 

func (n *DAGNode[T]) WithRecover() *DAGNode[T]

func (*DAGNode[T]) WithRetry added in v0.6.0 

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

func (*DAGNode[T]) WithTimeout added in v0.6.0 

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

type DAGPlan added in v0.4.0 

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

func NewDAG added in v0.4.0 

func NewDAG[T context.Context](nodes ...*DAGNode[T]) *DAGPlan[T]
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 added in v0.4.0 

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

func (*DAGPlan[T]) ExecNWithReport added in v0.6.0 

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

func (*DAGPlan[T]) ExecWithReport added in v0.6.0 

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

func (*DAGPlan[T]) Step added in v0.4.0 

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

func (*DAGPlan[T]) StepN added in v0.4.0 

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

func (*DAGPlan[T]) StepNWithReport added in v0.6.0 

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

func (*DAGPlan[T]) StepWithReport added in v0.6.0 

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

func (*DAGPlan[T]) ToDOT added in v0.4.0 

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

func (*DAGPlan[T]) ToMermaid added in v0.4.0 

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

func (*DAGPlan[T]) Validate added in v0.5.0 

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

type DAGReport added in v0.6.0 

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

func (*DAGReport) Failed added in v0.6.0 

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

func (*DAGReport) Node added in v0.6.0 

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

func (*DAGReport) Skipped added in v0.6.0 

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

func (*DAGReport) Successful added in v0.6.0 

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

type Middleware 

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

type NodeReport added in v0.6.0 

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

type NodeStatus added in v0.6.0 

type NodeStatus string
const (
	NodeStatusSuccess NodeStatus = "SUCCESS"
	NodeStatusFailed  NodeStatus = "FAILED"
	NodeStatusSkipped NodeStatus = "SKIPPED"
)

type Step 

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

func Branch added in v0.3.0 

func Branch[T context.Context](condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]
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]
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]

func DAG 

func DAG[T context.Context](nodes ...*DAGNode[T]) Step[T]
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]
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]

func DAGN 

func DAGN[T context.Context](limit int, nodes ...*DAGNode[T]) Step[T]
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 Each 

func Each[T context.Context, V any](seq iter.Seq[V], step func(ctx T, item V) error) Step[T]
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]

func Go 

func Go[T context.Context](steps ...Step[T]) Step[T]
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]
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]
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 added in v0.3.0 

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]
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 added in v0.3.0 

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]
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 added in v0.3.0 

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]

func Race 

func Race[T context.Context](steps ...Step[T]) Step[T]
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]
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 added in v0.3.0 

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

func (Step[T]) Catch 

func (s Step[T]) Catch(handler func(ctx T, err error) error) Step[T]

func (Step[T]) Exec 

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

func (Step[T]) Fallback 

func (s Step[T]) Fallback(fallback Step[T]) Step[T]
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]

func (Step[T]) GoN 

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

func (Step[T]) Once 

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

func (Step[T]) Race 

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

func (Step[T]) Recover 

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

func (Step[T]) Retry 

func (s Step[T]) Retry(attempts int, delay time.Duration) Step[T]
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]) Then 

func (s Step[T]) Then(steps ...Step[T]) Step[T]
Example 
package main

import (
	"context"
	"fmt"

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

func main() {
	s1 := flow.Step[context.Context](func(ctx context.Context) error {
		fmt.Println("First")
		return nil
	})
	s2 := flow.Step[context.Context](func(ctx context.Context) error {
		fmt.Println("Second")
		return nil
	})

	chained := s1.Then(s2)
	_ = chained.Exec(context.Background())
}

func (Step[T]) Timeout 

func (s Step[T]) Timeout(d time.Duration) Step[T]
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]

func (Step[T]) When 

func (s Step[T]) When(predicate func(ctx T) bool) Step[T]
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]
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())
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.