workflow

Documentation

Index

• Variables
• type Config
  • func NewConfig(provider config.Provider) (Config, error)
  • func NewDefaultConfig() Config
• type Connector
  • func NewConnector() *Connector
  • func (c *Connector) Add(nodeID string, o *graph.IO)
  • func (c *Connector) Connect(ctx context.Context) rxgo.Observable
  • func (c *Connector) Dispose()
  • func (c *Connector) Get(nodeID string) (*graph.IO, bool)
• type EdgeBuilder
• type Logger
• type Manager
  • func NewManager(wm *WorkflowManager) *Manager
  • func (m *Manager) Id() string
  • func (m *Manager) Start(ctx context.Context) (context.Context, *Connector)
  • func (m *Manager) Stop() error
• type ManagerBuilder
  • func NewManagerBuilder(wm *WorkflowManager) *ManagerBuilder
  • func (m *ManagerBuilder) Build() (*Manager, error)
  • func (m *ManagerBuilder) WithReq(req *gen.RunWorkflowRequest) *ManagerBuilder
• type ManagerType
• type MemoryLogger
  • func (m MemoryLogger) Debug(msg string, keyvals ...interface{})
  • func (m MemoryLogger) Error(msg string, keyvals ...interface{})
  • func (m MemoryLogger) Info(msg string, keyvals ...interface{})
  • func (m MemoryLogger) Warn(msg string, keyvals ...interface{})
• type NodeBuilder
• type RunningWorkflow
• type Workflow
  • func (w *Workflow) EnumerateProviders() []*gen.EnumeratedProvider
  • func (w *Workflow) GetNode(id string) (graph.Node, error)
  • func (w *Workflow) GetNodeInfo(n graph.Node) (*graph.Info, error)
  • func (w *Workflow) GetNodeProvider(id string) (graph.Node, error)
  • func (w *Workflow) WireNodes(ctx context.Context, nodeID string, input rxgo.Observable, manager *Manager) (*graph.IO, error)
• type WorkflowBuilder
  • func Default() *WorkflowBuilder[*gen.Node, *gen.Edge]
  • func NewBuilder[T graph.ProtoNode, U graph.ProtoEdge]() *WorkflowBuilder[T, U]
  • func (w *WorkflowBuilder[T, U]) Build() (*Workflow, error)
  • func (w *WorkflowBuilder[T, U]) WithBuiltEdges(edges ...graph.Edge) *WorkflowBuilder[T, U]
  • func (w *WorkflowBuilder[T, U]) WithBuiltNodes(nodes ...graph.Node) *WorkflowBuilder[T, U]
  • func (w *WorkflowBuilder[T, U]) WithEdgeTypes(n U, builder EdgeBuilder[U]) *WorkflowBuilder[T, U]
  • func (w *WorkflowBuilder[T, U]) WithNodeTypes(n T, builder NodeBuilder[T]) *WorkflowBuilder[T, U]
  • func (w *WorkflowBuilder[T, U]) WithNodes(nodes ...T) *WorkflowBuilder[T, U]
  • func (w *WorkflowBuilder[T, U]) WithProtoProject(project graph.ProtoProject[T, U]) *WorkflowBuilder[T, U]
• type WorkflowManager
  • func NewWorkflowManager() *WorkflowManager
  • func (m *WorkflowManager) Start(ctx context.Context, id string, req *gen.RunWorkflowRequest) (context.Context, *Connector)
  • func (m *WorkflowManager) Stop(id string) error
  • func (m *WorkflowManager) Traces() []*gen.WorkflowTrace

Variables

var ProviderSet = wire.NewSet(
	NewManagerBuilder,
	NewWorkflowManager,
)

Types

type Config

type Config struct {
	ManagerType ManagerType `yaml:"manager"`
}

func NewConfig

func NewConfig(provider config.Provider) (Config, error)

func NewDefaultConfig

func NewDefaultConfig() Config

type Connector

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

func NewConnector

func NewConnector() *Connector

func (*Connector) Add

func (c *Connector) Add(nodeID string, o *graph.IO)

func (*Connector) Connect

func (c *Connector) Connect(ctx context.Context) rxgo.Observable

func (*Connector) Dispose

func (c *Connector) Dispose()

func (*Connector) Get

func (c *Connector) Get(nodeID string) (*graph.IO, bool)

type EdgeBuilder

type EdgeBuilder[U graph.ProtoEdge] func(message U) graph.Edge

EdgeBuilder is a function that builds an edge from a proto message.

type Logger

type Logger interface {
	Debug(msg string, keyvals ...interface{})
	Info(msg string, keyvals ...interface{})
	Warn(msg string, keyvals ...interface{})
	Error(msg string, keyvals ...interface{})
}

type Manager

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

func NewManager

func NewManager(wm *WorkflowManager) *Manager

func (*Manager) Id

func (m *Manager) Id() string

func (*Manager) Start

func (m *Manager) Start(ctx context.Context) (context.Context, *Connector)

func (*Manager) Stop

func (m *Manager) Stop() error

type ManagerBuilder

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

func NewManagerBuilder

func NewManagerBuilder(wm *WorkflowManager) *ManagerBuilder

func (*ManagerBuilder) Build

func (m *ManagerBuilder) Build() (*Manager, error)

func (*ManagerBuilder) WithReq

func (m *ManagerBuilder) WithReq(req *gen.RunWorkflowRequest) *ManagerBuilder

type ManagerType

type ManagerType string

const (
	MemoryManagerType   ManagerType = "memory"
	TemporalManagerType ManagerType = "temporal"
)

type MemoryLogger

type MemoryLogger struct{}

func (MemoryLogger) Debug

func (m MemoryLogger) Debug(msg string, keyvals ...interface{})

func (MemoryLogger) Error

func (m MemoryLogger) Error(msg string, keyvals ...interface{})

func (MemoryLogger) Info

func (m MemoryLogger) Info(msg string, keyvals ...interface{})

func (MemoryLogger) Warn

func (m MemoryLogger) Warn(msg string, keyvals ...interface{})

type NodeBuilder

type NodeBuilder[T graph.ProtoNode] func(message T) graph.Node

NodeBuilder is a function that builds a node from a proto message.

type RunningWorkflow

type RunningWorkflow struct {
	Trace  *gen.WorkflowTrace
	Ctx    context.Context
	Cancel context.CancelFunc
	// contains filtered or unexported fields
}

type Workflow

type Workflow struct {
	ID        string
	ProjectID string
	Graph     graphlib.Graph[string, string]
	// TODO breadchris this should be a deterministic value based on the workflow node slice
	NodeLookup map[string]graph.Node
	EdgeLookup map[string]graph.Edge
}

Workflow is a directed graph of nodes that represent a workflow. The builder interface is immutable allowing for extensions.

func (*Workflow) EnumerateProviders

func (w *Workflow) EnumerateProviders() []*gen.EnumeratedProvider

func (*Workflow) GetNode

func (w *Workflow) GetNode(id string) (graph.Node, error)

func (*Workflow) GetNodeInfo

func (w *Workflow) GetNodeInfo(n graph.Node) (*graph.Info, error)

TODO breadchris separate "infer" and "collect" type information TODO breadchris node.Type should be passed workflow since a node's info can depend on other nodes

func (*Workflow) GetNodeProvider

func (w *Workflow) GetNodeProvider(id string) (graph.Node, error)

func (*Workflow) WireNodes

func (w *Workflow) WireNodes(
	ctx context.Context,
	nodeID string,
	input rxgo.Observable,
	manager *Manager,
) (*graph.IO, error)

WireNodes wires the nodes in the workflow together and returns an observable that can be subscribed to. Nodes are executed when an event is received on the input observable.

type WorkflowBuilder

type WorkflowBuilder[T graph.ProtoNode, U graph.ProtoEdge] struct {
	// contains filtered or unexported fields
}

WorkflowBuilder is a builder for a workflow. It is immutable allowing for extensions.

func Default

func Default() *WorkflowBuilder[*gen.Node, *gen.Edge]

func NewBuilder

func NewBuilder[T graph.ProtoNode, U graph.ProtoEdge]() *WorkflowBuilder[T, U]

func (*WorkflowBuilder[T, U]) Build

func (w *WorkflowBuilder[T, U]) Build() (*Workflow, error)

func (*WorkflowBuilder[T, U]) WithBuiltEdges

func (w *WorkflowBuilder[T, U]) WithBuiltEdges(edges ...graph.Edge) *WorkflowBuilder[T, U]

func (*WorkflowBuilder[T, U]) WithBuiltNodes

func (w *WorkflowBuilder[T, U]) WithBuiltNodes(nodes ...graph.Node) *WorkflowBuilder[T, U]

func (*WorkflowBuilder[T, U]) WithEdgeTypes

func (w *WorkflowBuilder[T, U]) WithEdgeTypes(n U, builder EdgeBuilder[U]) *WorkflowBuilder[T, U]

func (*WorkflowBuilder[T, U]) WithNodeTypes

func (w *WorkflowBuilder[T, U]) WithNodeTypes(n T, builder NodeBuilder[T]) *WorkflowBuilder[T, U]

func (*WorkflowBuilder[T, U]) WithNodes

func (w *WorkflowBuilder[T, U]) WithNodes(nodes ...T) *WorkflowBuilder[T, U]

func (*WorkflowBuilder[T, U]) WithProtoProject

func (w *WorkflowBuilder[T, U]) WithProtoProject(project graph.ProtoProject[T, U]) *WorkflowBuilder[T, U]

type WorkflowManager

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

func NewWorkflowManager

func NewWorkflowManager() *WorkflowManager

func (*WorkflowManager) Start

func (m *WorkflowManager) Start(ctx context.Context, id string, req *gen.RunWorkflowRequest) (context.Context, *Connector)

func (*WorkflowManager) Stop

func (m *WorkflowManager) Stop(id string) error

func (*WorkflowManager) Traces

func (m *WorkflowManager) Traces() []*gen.WorkflowTrace