flow

package module

Go to main page

Versions in this module

v0

v0.8.1

Sep 12, 2026

v0.8.0

Sep 6, 2026

Changes in this version

type Middleware

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

v0.7.1

Sep 6, 2026

v0.7.0

Sep 6, 2026

Changes in this version

type Step

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

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

v0.6.0

Sep 5, 2026

Changes in this version

+ func DAGEdgesNWithReport[T context.Context](limit int, connections ...DAGConnection[T]) func(ctx T) (*DAGReport, 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 DAGWithReport[T context.Context](nodes ...*DAGNode[T]) func(ctx T) (*DAGReport, error)

type DAGNode

+ 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]) WithCatch(handler func(ctx T, err error) error) *DAGNode[T]

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

+ func (n *DAGNode[T]) WithRecover() *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 (p *DAGPlan[T]) ExecNWithReport(limit int, ctx T) (*DAGReport, error)

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

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

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

+ type DAGReport struct

+ Duration time.Duration

+ EndTime time.Time

+ Nodes []NodeReport

+ StartTime time.Time

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

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

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

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

+ type NodeReport struct

+ Duration time.Duration

+ EndTime time.Time

+ Err error

+ Name string

+ StartTime time.Time

+ Status NodeStatus

+ type NodeStatus string

+ const NodeStatusFailed

+ const NodeStatusSkipped

+ const NodeStatusSuccess

v0.5.0

Sep 5, 2026

Changes in this version

+ var ErrDAGCycle = errors.New("cycle detected in DAG")

+ var ErrDAGDuplicateNode = errors.New("duplicate node name in DAG")

+ var ErrDAGEmptyNodeName = errors.New("empty node name in DAG")

+ var ErrDAGNilNode = errors.New("nil node in DAG")

+ var ErrDAGUnknownDependency = errors.New("unknown dependency in DAG")

type DAGPlan

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

v0.4.0

Sep 5, 2026

Changes in this version

+ 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)

+ type DAGPlan struct

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

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

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

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

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

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

v0.3.0

Sep 5, 2026

Changes in this version

+ 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 Step

+ func Branch[T context.Context](condition func(ctx T) bool, ifBranch, elseBranch Step[T]) 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 PipeSeq[T context.Context, In, Out any](seq iter.Seq[In], transform func(ctx T, item In) (Out, error), ...) 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 (s Step[T]) Branch(condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]

v0.1.0

Sep 5, 2026

Changes in this version

+ type DAGConnection struct

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

+ type DAGEdgeBuilder struct

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

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

+ type DAGNode struct

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

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

+ type Middleware func(next Step[T]) Step[T]

+ type Step func(ctx T) error

+ 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 Chunk[T context.Context, V any](seq iter.Seq[V], size int, step func(ctx T, batch []V) error) Step[T]

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

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

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

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

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

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

+ func Once[T context.Context](step Step[T]) 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]) 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]

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL