pdag

Documentation

Overview

Package pdag allows to define a DAG of processing fuctions.

The processing can be triggered at any node of the graph and will follow the directed edges of the graph. Before a function in a node is executed all its parents have to finish execution. Thus the graph defines the order in which the functions are executed and which functions are executed serially or in parallel.

A single shared object (of type interface{}) is passed to all functions. It's the responsibility of the functions to coordinate synchronized access to the object.

If an error occurs in any functions all processing seizes and the error is returned by the Trigger function.

For example

root := NewNodeWithParents(a)
NewNodeWithParents(d, root.Child(b), root.Child(c))

state := map[string]string{}
root.Trigger(data)

defines a diamond-shaped DAG. Execution starts at the root and after function 'a', functions 'b' and 'c' will be executed in parallel. Once both have completed, function 'd' will be called. All will be passed the value of 'state'.

An instance of processing DAG is thread-safe. Data consistency has to be ensured in the shared state.

Index

• func NoOp(_ context.Context, _ interface{}) error
• type Node
  • func NewNodeWithParents(fn ProcessFn, parents ...*Node) *Node
  • func (n *Node) Child(fn ProcessFn) *Node
  • func (n *Node) Trigger(ctx context.Context, state interface{}) error
• type ProcessFn

Functions

func NoOp

func NoOp(_ context.Context, _ interface{}) error

NoOp does nothing.

Types

type Node

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

Node of the Dag.

func NewNodeWithParents

func NewNodeWithParents(fn ProcessFn, parents ...*Node) *Node

NewNodeWithParents creates a new Node in the processing DAG. It takes the function to be executed in this node and an optional list of parent nodes.

func (*Node) Child

func (n *Node) Child(fn ProcessFn) *Node

Child is a shorthand function that creates a child node of an existing node.

func (*Node) Trigger

func (n *Node) Trigger(ctx context.Context, state interface{}) error

Trigger starts execution at the current node and executes all functions that are descendents of this node. It blocks until all nodes have been executed. If any of the functions returns an error, execution ceases and the error is returned. Note: Trigger can be called on any node in the graph and will only call the descendants of that node.

type ProcessFn

type ProcessFn func(ctx context.Context, state interface{}) error

ProcessFn is the type of the processing function for each node.