dag

README

dag has two main concept:

1. Pipeline executes the functions sequentially and in order.
2. Spawns executes the functions concurrently, so there is no ordering guarantee.

Example 1

d := dag.New()
d.Pipeline(f1, f2, f3)
d.Run()

In the above example, f1 starts first, and after completion, f2 starts then f3.
Full example : examples/ex1/ex1.go

Example 2

d := dag.New()
d.Spawns(f1, f2, f3)
d.Run()

The order of execution of f1, f2 and f3 is nondeterministic
Full example : examples/ex2/ex2.go

Example 3

In this example f4 must be executed after complition of f1, f2 and f3. You can use Join method:

d := dag.New()
d.Spawns(f1, f2, f3).Join().Pipeline(f4)
d.Run()

Full example : examples/ex3/ex3.go

Example 4

After pipeline we can use Then method:

d := dag.New()
d.Pipeline(f1, f2, f3).Then().Spawns(f4, f5, f6)
d.Run()

Full example : examples/ex4/ex4.go

Example 5

d := dag.New()
d.Spawns(f1, f2, f3).
	Join().
	Pipeline(f4, f5).
	Then().
	Spawns(f6, f7, f8)
d.Run()

Full example : examples/ex5/ex5.go

Example 6

We want to execute two pipeline concrrently, we can use pipeline.Of inside the Spawns method:

d := dag.New()
d.Spawns(pipeline.Of(f1, f3), pipeline.Of(f2, f4)).
	Join().
	Pipeline(f5)
d.Run()

Full example : examples/ex6/ex6.go

Example 7

We can use OnComplete method after Pipeline or Spawns to notify when functions has completed.

d := dag.New()
d.Pipeline(f1, f2).OnComplete(f3).
	  Then().
  Spawns(f1, f2).OnComplete(f4)
d.Run()

Full example : examples/ex7/ex7.go

Example 8

Basically, Run() will block until all functions are done. If you don't want to be blocked, you can use RunAsync() method. It accepts a callback function, that will be called when all functions are done.

d := dag.New()
d.Pipeline(f1, f2).Then().Spawns(f3, f4)
d.RunAsync(onComplete)

Full example : examples/ex8/ex8.go

Documentation

Index

• type Codec
• type Context
  • func GetContext() *Context
• type Dag
  • func New() *Dag
  • func (dag *Dag) Any(tasks ...func()) *anyResult
  • func (dag *Dag) Parse(dsl string) *Dag
  • func (dag *Dag) Pipeline(tasks ...func()) *pipelineResult
  • func (dag *Dag) Run()
  • func (dag *Dag) RunAsync(onComplete func())
  • func (dag *Dag) Spawns(tasks ...func()) *spawnsResult
• type FetchPipeline
• type Input
• type Job
• type Output
• type Processor

Types

type Codec

type Codec interface {
	Encode(in []byte) []byte
	Decode(in []byte) []byte
}

type Context

type Context struct {
	Channel chan []byte
	Codec   Codec
}

func GetContext

func GetContext() *Context

type Dag

type Dag struct {
	// contains filtered or unexported fields
}

Dag represents directed acyclic graph

func New

func New() *Dag

New creates new DAG

func (*Dag) Any

func (dag *Dag) Any(tasks ...func()) *anyResult

func (*Dag) Parse

func (dag *Dag) Parse(dsl string) *Dag

func (*Dag) Pipeline

func (dag *Dag) Pipeline(tasks ...func()) *pipelineResult

Pipeline executes tasks sequentially

func (*Dag) Run

func (dag *Dag) Run()

Run starts the tasks It will block until all functions are done

func (*Dag) RunAsync

func (dag *Dag) RunAsync(onComplete func())

RunAsync executes Run on another goroutine

func (*Dag) Spawns

func (dag *Dag) Spawns(tasks ...func()) *spawnsResult

Spawns executes tasks concurrently

type FetchPipeline

type FetchPipeline struct {
	Inputs []Input

	Processors []Processor

	Outputs []Output
}

type Input

type Input interface {
	Read()
}

type Job

type Job struct {
	// contains filtered or unexported fields
}

Job - Each job consists of one or more tasks Each Job can runs tasks in order(Sequential) or unordered

type Output

type Output interface {
	Write()
}

type Processor

type Processor interface {
}