activity

Documentation

Index

• func ApplyTaskDataOutput(element schema.BaseElementInterface, dataOutputs map[string]any) map[string]data.IItem
• func FetchTaskDataInput(locator data.IFlowDataLocator, element schema.BaseElementInterface) (headers, dataSets, dataObjects map[string]any)
• type ActiveBoundaryTrace
  • func (b ActiveBoundaryTrace) TraceInterface()
• type ActiveTaskTrace
• type Activity
• type Constructor
• type Harness
  • func NewHarness(ctx context.Context, wiring *flow_node.Wiring, element *schema.FlowNode, ...) (node *Harness, err error)
  • func (node *Harness) Activity() Activity
  • func (node *Harness) ConsumeEvent(ev event.IEvent) (result event.ConsumptionResult, err error)
  • func (node *Harness) Element() schema.FlowNodeInterface
  • func (node *Harness) NextAction(flow flow_interface.T) chan flow_node.IAction
  • func (node *Harness) RegisterEventConsumer(consumer event.IConsumer) (err error)

Functions

func ApplyTaskDataOutput

func ApplyTaskDataOutput(element schema.BaseElementInterface, dataOutputs map[string]any) map[string]data.IItem

func FetchTaskDataInput

func FetchTaskDataInput(locator data.IFlowDataLocator, element schema.BaseElementInterface) (headers, dataSets, dataObjects map[string]any)

Types

type ActiveBoundaryTrace

type ActiveBoundaryTrace struct {
	Start bool
	Node  schema.FlowNodeInterface
}

func (ActiveBoundaryTrace) TraceInterface

func (b ActiveBoundaryTrace) TraceInterface()

type ActiveTaskTrace

type ActiveTaskTrace interface {
	tracing.ITrace
	Execute()
}

ActiveTaskTrace describes common channel handler for all tasks

type Activity

type Activity interface {
	flow_node.IFlowNode
	// Cancel initiates a cancellation of activity and returns a channel
	// that will signal a boolean (`true` if cancellation was successful,
	// `false` otherwise)
	Cancel() <-chan bool
}

Activity is a generic interface to flow nodes that are activities

type Constructor

type Constructor = func(*flow_node.Wiring) (node Activity, err error)

type Harness

type Harness struct {
	*flow_node.Wiring
	// contains filtered or unexported fields
}

func NewHarness

func NewHarness(ctx context.Context,
	wiring *flow_node.Wiring,
	element *schema.FlowNode,
	idGenerator id.IGenerator,
	constructor Constructor,
) (node *Harness, err error)

func (*Harness) Activity

func (node *Harness) Activity() Activity

func (*Harness) ConsumeEvent

func (node *Harness) ConsumeEvent(ev event.IEvent) (result event.ConsumptionResult, err error)

func (*Harness) Element

func (node *Harness) Element() schema.FlowNodeInterface

func (*Harness) NextAction

func (node *Harness) NextAction(flow flow_interface.T) chan flow_node.IAction

func (*Harness) RegisterEventConsumer

func (node *Harness) RegisterEventConsumer(consumer event.IConsumer) (err error)