dag

package

Go to main page

Versions in this module

v0

v0.0.9

Jan 19, 2025

v0.0.8

Dec 26, 2024

v0.0.7

Nov 30, 2024

Changes in this version

+ type Direction string

+ const LR

+ const TB

v0.0.6

Nov 26, 2024

v0.0.5

Nov 24, 2024

v0.0.4

Nov 24, 2024

Changes in this version

+ func ParseFormToMap(body []byte, data *map[string]any) error

v0.0.3

Nov 23, 2024

Changes in this version

type NodeType

+ func (c NodeType) String() string

v0.0.2

Nov 23, 2024

Changes in this version

+ const ContextIndex

+ const DefaultChannelSize

+ const Delimiter

+ const RetryInterval

+ func ClearDAG()

+ type Context struct

+ Query map[string]any

+ func UserContext(ctx context.Context) *Context

+ func (ctx *Context) Get(key string) string

type DAG

+ func (tm *DAG) GetLastNodes() ([]*Node, error)

+ func (tm *DAG) GetNextNodes(key string) ([]*Node, error)

+ func (tm *DAG) GetPreviousNodes(key string) ([]*Node, error)

+ func (tm *DAG) IsLastNode(key string) (bool, error)

type Node

+ ID string

+ Label string

+ NodeType NodeType

+ type NodeType int

+ const Function

+ const Page

+ func (c NodeType) IsValid() bool

type TaskManager

+ func (tm *TaskManager) ProcessTask(ctx context.Context, startNode string, payload json.RawMessage)

+ func (tm *TaskManager) Stop()

+ type TaskState struct

+ NodeID string

+ Result mq.Result

+ Status mq.Status

+ UpdatedAt time.Time

v0.0.1

Nov 2, 2024

Changes in this version

+ func AddDAG(key string, handler *DAG)

+ func AddHandler(key string, handler func(string) mq.Processor)

+ func AvailableDAG() []string

+ func AvailableHandlers() []string

+ func GetHandler(key string) func(string) mq.Processor

+ func GetVal(c context.Context, v string, data map[string]any) (key string, val any)

+ func Header(c context.Context, headerKey string) (val map[string]any, exists bool)

+ func HeaderVal(c context.Context, headerKey string, key string) (val any)

+ func WsEvents(s *sio.Server)

+ type Condition interface

+ Match func(data any) bool

+ type ConditionProcessor interface

+ SetConditions func(map[string]Condition)

+ type DAG struct

+ Error error

+ Notifier *sio.Server

+ func GetDAG(key string) *DAG

+ func NewDAG(name, key string, opts ...mq.Option) *DAG

+ func (tm *DAG) AddCondition(fromNode FromNode, conditions map[When]Then) *DAG

+ func (tm *DAG) AddDAGNode(name string, key string, dag *DAG, firstNode ...bool)

+ func (tm *DAG) AddDeferredNode(name, key string, firstNode ...bool) error

+ func (tm *DAG) AddEdge(label, from string, targets ...string) *DAG

+ func (tm *DAG) AddIterator(label, from string, targets ...string) *DAG

+ func (tm *DAG) AddNode(name, key string, handler mq.Processor, firstNode ...bool) *DAG

+ func (tm *DAG) AssignTopic(topic string)

+ func (tm *DAG) ClassifyEdges(startNodes ...string) (string, bool, error)

+ func (tm *DAG) Close() error

+ func (tm *DAG) Consume(ctx context.Context) error

+ func (tm *DAG) ExportDOT() string

+ func (tm *DAG) GetKey() string

+ func (tm *DAG) GetReport() string

+ func (tm *DAG) GetStartNode() string

+ func (tm *DAG) GetType() string

+ func (tm *DAG) Handlers()

+ func (tm *DAG) IsReady() bool

+ func (tm *DAG) Pause(_ context.Context) error

+ func (tm *DAG) PauseConsumer(ctx context.Context, id string)

+ func (tm *DAG) PrintGraph()

+ func (tm *DAG) Process(ctx context.Context, payload []byte) mq.Result

+ func (tm *DAG) ProcessTask(ctx context.Context, task *mq.Task) mq.Result

+ func (tm *DAG) Publish(w http.ResponseWriter, r *http.Request)

+ func (tm *DAG) ReportNodeResult(callback func(mq.Result))

+ func (tm *DAG) Request(w http.ResponseWriter, r *http.Request)

+ func (tm *DAG) Resume(_ context.Context) error

+ func (tm *DAG) ResumeConsumer(ctx context.Context, id string)

+ func (tm *DAG) SaveDOTFile(filename string) error

+ func (tm *DAG) SavePNG(pngFile string) error

+ func (tm *DAG) SaveSVG(svgFile string) error

+ func (tm *DAG) Schedule(w http.ResponseWriter, r *http.Request)

+ func (tm *DAG) ScheduleTask(ctx context.Context, payload []byte, opts ...mq.SchedulerOption) mq.Result

+ func (tm *DAG) SetKey(key string)

+ func (tm *DAG) SetNotifyResponse(callback mq.Callback)

+ func (tm *DAG) SetStartNode(node string)

+ func (tm *DAG) SetupWS() *sio.Server

+ func (tm *DAG) Start(ctx context.Context, addr string) error

+ func (tm *DAG) Stop(ctx context.Context) error

+ func (tm *DAG) TopologicalSort() (stack []string)

+ func (tm *DAG) Validate() error

+ type Edge struct

+ From *Node

+ Label string

+ To []*Node

+ Type EdgeType

+ type EdgeType int

+ const Iterator

+ const Simple

+ func (c EdgeType) IsValid() bool

+ type FromNode string

+ type List struct

+ Handlers map[string]*DAG

+ type Node struct

+ Edges []Edge

+ Key string

+ Name string

+ func (n *Node) Close() error

+ func (n *Node) ProcessTask(ctx context.Context, msg *mq.Task) mq.Result

+ type Operation struct

+ GeneratedFields []string

+ ID string

+ Key string

+ OptionalFields []string

+ Payload Payload

+ RequiredFields []string

+ Type string

+ func (e *Operation) Close() error

+ func (e *Operation) Consume(_ context.Context) error

+ func (e *Operation) GetKey() string

+ func (e *Operation) GetType() string

+ func (e *Operation) Pause(_ context.Context) error

+ func (e *Operation) ProcessTask(_ context.Context, task *mq.Task) mq.Result

+ func (e *Operation) Resume(_ context.Context) error

+ func (e *Operation) SetConfig(payload Payload)

+ func (e *Operation) SetKey(key string)

+ func (e *Operation) Stop(_ context.Context) error

+ func (e *Operation) ValidateFields(c context.Context, payload []byte) (map[string]any, error)

+ type Operations struct

+ Handlers map[string]func(string) mq.Processor

+ type Payload struct

+ Data map[string]any

+ GeneratedFields []string

+ Mapping map[string]string

+ Providers []Provider

+ type Processor interface

+ SetConfig func(Payload)

+ type Provider struct

+ Database string

+ Defaults map[string]any

+ InsertMapping map[string]any

+ Mapping map[string]any

+ ProviderType string

+ Query string

+ Source string

+ UpdateMapping map[string]any

+ type Request struct

+ Interval time.Duration

+ Overlap bool

+ Payload json.RawMessage

+ Recurring bool

+ Schedule bool

+ type TaskManager struct

+ func NewTaskManager(d *DAG, taskID string) *TaskManager

+ type Then string

+ type WaitGroup struct

+ func NewWaitGroup() *WaitGroup

+ func (awg *WaitGroup) Add(delta int)

+ func (awg *WaitGroup) Done()

+ func (awg *WaitGroup) Reset()

+ func (awg *WaitGroup) Wait()

+ type When string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL