flow

README

go-flow

Documentation

Index

• 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]
• type Middleware
• type Step
  • 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 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]

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]

type Middleware

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

type Step

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

func Chunk

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

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]

func DAGEdges

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

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]

func Each

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

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]

func GoN

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

func Once

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

func Race

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

func Seq

func Seq[T context.Context](steps ...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]

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]

func (Step[T]) Then

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

func (Step[T]) Timeout

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

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]

func (Step[T]) Wrap

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