flow

Documentation

Overview

flow/executor.go

Index

• func GetCompensation(effect effectus.Effect) string
• func Run(program *Program, executor effectus.Executor) (interface{}, error)
• func RunContext(ctx context.Context, program *Program, executor effectus.Executor) (interface{}, error)
• type CompiledFlow
  • func (cf *CompiledFlow) GetPriority() int
• type Compiler
  • func (c *Compiler) CompileFile(path string, schema effectus.SchemaInfo) (effectus.Spec, error)
  • func (c *Compiler) CompileFiles(paths []string, schema effectus.SchemaInfo) (effectus.Spec, error)
  • func (c *Compiler) CompileParsedFile(file *ast.File, path string, schema effectus.SchemaInfo) (effectus.Spec, error)
• type Executor
  • func NewExecutor(verbRegistry common.VerbRegistry, options ...ExecutorOption) *Executor
  • func (fe *Executor) ExecuteProgram(ctx context.Context, flowName string, program *Program, facts common.Facts) (interface{}, error)
• type ExecutorOption
  • func WithCapabilitySystem(capSystem *capability.CapabilitySystem) ExecutorOption
  • func WithSaga(store schema.SagaStore) ExecutorOption
• type Program
  • func Atomic(name string, program *Program) *Program
  • func Do(effect effectus.Effect, cont func(interface{}) *Program) *Program
  • func DoWithCompensation(effect effectus.Effect, compensation string, cont func(interface{}) *Program) *Program
  • func Error(err error) *Program
  • func FromList(effects []effectus.Effect) *Program
  • func FromListWithCompensation(effects []effectus.Effect, compensations map[string]string) *Program
  • func Pure(value interface{}) *Program
  • func Transaction(sagaID, name string, program *Program) *Program
  • func (p *Program) Bind(f func(interface{}) *Program) *Program
  • func (p *Program) FlatMap(f func(interface{}) *Program) *Program
  • func (p *Program) IsTransactional() bool
  • func (p *Program) Map(f func(interface{}) interface{}) *Program
  • func (p *Program) Then(next *Program) *Program
  • func (p *Program) ToAtomic(name string) *Program
  • func (p *Program) ToTransaction(sagaID, name string) *Program
  • func (p *Program) WithCompensation(getCompensation func(verb string) string) *Program
• type ProgramTag
• type Spec
  • func (s *Spec) Execute(ctx context.Context, facts effectus.Facts, ex effectus.Executor) error
  • func (s *Spec) GetName() string
  • func (s *Spec) RequiredFacts() []string
• type TransactionInfo
  • func ExtractTransactions(program *Program) []*TransactionInfo

Functions

func GetCompensation

func GetCompensation(effect effectus.Effect) string

GetCompensation extracts compensation information from an effect

func Run

func Run(program *Program, executor effectus.Executor) (interface{}, error)

Run executes a program through the legacy context-free API.

func RunContext

func RunContext(ctx context.Context, program *Program, executor effectus.Executor) (interface{}, error)

RunContext executes a program and propagates cancellation to every effect.

Types

type CompiledFlow

type CompiledFlow struct {
	Name       string
	Priority   int
	Predicates []*schema.Predicate
	Program    *Program
	FactPaths  []string
	SourceFile string
}

CompiledFlow represents a flow after compilation

func (*CompiledFlow) GetPriority

func (cf *CompiledFlow) GetPriority() int

GetPriority implements the common.Prioritized interface

type Compiler

type Compiler struct{}

Compiler implements the Compiler interface for flow-style rules

func (*Compiler) CompileFile

func (c *Compiler) CompileFile(path string, schema effectus.SchemaInfo) (effectus.Spec, error)

CompileFile compiles a rule file into a flow-style spec

func (*Compiler) CompileFiles

func (c *Compiler) CompileFiles(paths []string, schema effectus.SchemaInfo) (effectus.Spec, error)

CompileFiles compiles multiple rule files into a single flow-style spec

func (*Compiler) CompileParsedFile

func (c *Compiler) CompileParsedFile(file *ast.File, path string, schema effectus.SchemaInfo) (effectus.Spec, error)

CompileParsedFile compiles a parsed file into a flow-style spec

type Executor

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

Executor is the main executor for flow programs with saga and capability support

func NewExecutor

func NewExecutor(verbRegistry common.VerbRegistry, options ...ExecutorOption) *Executor

NewExecutor creates a new executor for flow programs

func (*Executor) ExecuteProgram

func (fe *Executor) ExecuteProgram(ctx context.Context, flowName string, program *Program, facts common.Facts) (interface{}, error)

ExecuteProgram executes a flow program with saga and capability support

type ExecutorOption

type ExecutorOption func(*Executor)

ExecutorOption defines an option for configuring the flow executor

func WithCapabilitySystem

func WithCapabilitySystem(capSystem *capability.CapabilitySystem) ExecutorOption

WithCapabilitySystem enables capability-based locking

func WithSaga

func WithSaga(store schema.SagaStore) ExecutorOption

WithSaga enables saga-style compensation for failed executions

type Program

type Program struct {
	Tag         ProgramTag
	Pure        interface{}
	Effect      effectus.Effect
	Transaction *TransactionInfo
	Continue    func(interface{}) *Program
}

Program represents a free monad over Effect with saga transaction support

func Atomic

func Atomic(name string, program *Program) *Program

Atomic creates an atomic transaction around a program

func Do

func Do(effect effectus.Effect, cont func(interface{}) *Program) *Program

Do creates a program that performs an effect then continues

func DoWithCompensation

func DoWithCompensation(effect effectus.Effect, compensation string, cont func(interface{}) *Program) *Program

DoWithCompensation creates a program that performs an effect with compensation info

func Error

func Error(err error) *Program

Error creates a program that immediately returns an error

func FromList

func FromList(effects []effectus.Effect) *Program

FromList converts a list of Effects to a Program This is the "canonical embedding" α from the theoretical foundation

func FromListWithCompensation

func FromListWithCompensation(effects []effectus.Effect, compensations map[string]string) *Program

FromListWithCompensation converts effects to a program with compensation

func Pure

func Pure(value interface{}) *Program

Pure creates a program that just returns a value

func Transaction

func Transaction(sagaID, name string, program *Program) *Program

Transaction creates a saga transaction boundary around a program

func (*Program) Bind

func (p *Program) Bind(f func(interface{}) *Program) *Program

Bind sequences two programs together (monadic bind)

func (*Program) FlatMap

func (p *Program) FlatMap(f func(interface{}) *Program) *Program

FlatMap is an alias for Bind with a more familiar name

func (*Program) IsTransactional

func (p *Program) IsTransactional() bool

IsTransactional returns true if the program contains any transaction boundaries

func (*Program) Map

func (p *Program) Map(f func(interface{}) interface{}) *Program

Map transforms the result value with a pure function

func (*Program) Then

func (p *Program) Then(next *Program) *Program

Then chains a program after this one, ignoring this program's result

func (*Program) ToAtomic

func (p *Program) ToAtomic(name string) *Program

ToAtomic wraps a program in an atomic transaction

func (*Program) ToTransaction

func (p *Program) ToTransaction(sagaID, name string) *Program

ToTransaction wraps a program in a saga transaction

func (*Program) WithCompensation

func (p *Program) WithCompensation(getCompensation func(verb string) string) *Program

WithCompensation adds compensation information to all effects in the program

type ProgramTag

type ProgramTag int

ProgramTag identifies the type of program node

const (
	// PureProgramTag is a pure value
	PureProgramTag ProgramTag = iota
	// EffectProgramTag is an effect with a continuation
	EffectProgramTag
	// TransactionProgramTag marks a saga transaction boundary
	TransactionProgramTag
)

type Spec

type Spec struct {
	Name         string
	Flows        []*CompiledFlow
	FactPaths    []string
	SagaEnabled  bool                         // Whether to use saga execution
	SagaStore    schema.SagaStore             // Saga store for transaction management
	CapSystem    *capability.CapabilitySystem // Capability system for locking
	VerbRegistry common.VerbRegistry          // Verb registry for execution
}

Spec implements the effectus.Spec interface for flow rules

func (*Spec) Execute

func (s *Spec) Execute(ctx context.Context, facts effectus.Facts, ex effectus.Executor) error

Execute runs all flows in the spec with saga and capability support

func (*Spec) GetName

func (s *Spec) GetName() string

GetName returns the name of this spec

func (*Spec) RequiredFacts

func (s *Spec) RequiredFacts() []string

RequiredFacts returns the list of fact paths required by this spec

type TransactionInfo

type TransactionInfo struct {
	SagaID       string   // Unique saga identifier
	Name         string   // Human-readable transaction name
	Compensation string   // Inverse verb for compensation
	Program      *Program // The program to execute within the transaction
	IsAtomic     bool     // Whether this transaction should be atomic
}

TransactionInfo holds metadata for saga transaction boundaries

func ExtractTransactions

func ExtractTransactions(program *Program) []*TransactionInfo

ExtractTransactions extracts all transaction boundaries from a program This is useful for saga executors to understand the transaction structure