Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewNodeContext ¶
func NewNodeContext(ctx context.Context, node *NodeContext) context.Context
NewNodeContext returns a new context with the given NodeContext.
Types ¶
type Checkpoint ¶
type Checkpoint struct {
ID string `json:"id"`
Received map[string]int `json:"received"`
Visited map[string]bool `json:"visited"`
State map[string]any `json:"state"`
}
Checkpoint captures the execution progress of a Task so it can be resumed. Use Clone() to create a deep copy if you need to modify the checkpoint.
func (*Checkpoint) Clone ¶
func (c *Checkpoint) Clone() *Checkpoint
Clone returns a deep copy of the checkpoint so callers can modify it without affecting the original snapshot.
type Checkpointer ¶
type Checkpointer interface {
Save(ctx context.Context, checkpoint *Checkpoint) error
Resume(ctx context.Context, checkpointID string) (*Checkpoint, error)
}
Checkpointer persists and restores checkpoints for a task identified by checkpointID. Save and Resume must be safe for concurrent use.
type CompileOption ¶
type CompileOption func(*compileConfig)
CompileOption configures Graph compilation.
func WithCheckpointer ¶
func WithCheckpointer(checkpointer Checkpointer) CompileOption
WithCheckpointer registers a Checkpointer used by the compiled Executor.
type EdgeCondition ¶
EdgeCondition is a function that determines if an edge should be followed based on the current state.
type EdgeOption ¶
type EdgeOption func(*conditionalEdge)
EdgeOption configures an edge before it is added to the graph.
func WithEdgeCondition ¶
func WithEdgeCondition(condition EdgeCondition) EdgeOption
WithEdgeCondition sets a condition that must return true for the edge to be taken.
type ExecuteOption ¶
type ExecuteOption func(*executeOptions)
ExecuteOption defines an option for the Execute method.
func WithCheckpointID ¶
func WithCheckpointID(CheckpointID string) ExecuteOption
WithCheckpointID sets a specific CheckpointID for the execution.
type Executor ¶
type Executor struct {
// contains filtered or unexported fields
}
Executor represents a compiled graph ready for execution. It is safe for concurrent use; each Execute call runs on an isolated execution context.
func NewExecutor ¶
func NewExecutor(g *Graph, checkpointer Checkpointer) *Executor
NewExecutor creates a new Executor for the given graph.
type Graph ¶
type Graph struct {
// contains filtered or unexported fields
}
Graph represents a directed graph of processing nodes. Cycles are rejected at compile time.
func (*Graph) AddEdge ¶
func (g *Graph) AddEdge(from, to string, opts ...EdgeOption) *Graph
AddEdge adds a directed edge from one node to another. Options can configure the edge. Returns the graph for chaining.
func (*Graph) AddNode ¶
AddNode adds a named node with its handler to the graph. Returns the graph for chaining.
func (*Graph) Compile ¶
func (g *Graph) Compile(opts ...CompileOption) (*Executor, error)
Compile validates and compiles the graph into an Executor. Nodes wait for all activated incoming edges to complete before executing (join semantics). An edge is "activated" when its source node executes and chooses that edge. Provide no WithCheckpointer option to disable checkpoint persistence.
func (*Graph) SetEntryPoint ¶
SetEntryPoint marks a node as the entry point. Returns the graph for chaining.
func (*Graph) SetFinishPoint ¶
SetFinishPoint marks a node as the finish point. Returns the graph for chaining.
type Middleware ¶
Middleware is a function that wraps a Handler with additional functionality.
func ChainMiddlewares ¶
func ChainMiddlewares(mws ...Middleware) Middleware
ChainMiddlewares composes middlewares into one, applying them in order. The first middleware becomes the outermost wrapper.
func Retry ¶
func Retry(attempts int, opts ...retry.Option) Middleware
Retry returns a middleware that retries node handlers with exponential backoff.
Parameters:
attempts: The total number of attempts to execute the handler, including the initial attempt.
For example, attempts=3 means up to 3 tries (1 initial + 2 retries).
opts: Optional configuration for retry behavior. See retry.Option (from github.com/go-kratos/kit/retry) for details.
Behavior:
- The same `state` value is passed to the handler on each attempt. Handlers must not mutate `state`.
- If all attempts are exhausted and the handler continues to return an error, the last error is returned and no further retries are performed.
- Retry behavior (e.g., backoff, which errors are retryable) can be customized via retry.Option.
Example usage:
// Retry up to 5 times with exponential backoff, only on specific errors.
mw := Retry(5,
retry.WithBackoff(retry.NewExponentialBackoff()),
retry.WithRetryable(func(err error) bool {
return errors.Is(err, ErrTemporary)
}),
)
type NodeContext ¶
type NodeContext struct {
Name string
}
NodeContext holds information about the current node in the graph.
func FromNodeContext ¶
func FromNodeContext(ctx context.Context) (*NodeContext, bool)
FromNodeContext retrieves the NodeContext from the context, if present.
type Option ¶
type Option func(*Graph)
Option configures the Graph behavior.
func WithMiddleware ¶
func WithMiddleware(ms ...Middleware) Option
WithMiddleware sets a global middleware applied to all node handlers.
func WithParallel ¶
WithParallel toggles parallel fan-out execution. Defaults to true.
type Task ¶
type Task struct {
// contains filtered or unexported fields
}
Task coordinates a single execution of the graph using a ready-queue based scheduler. This implementation combines: - Autogen's clean ready-queue + dependency counting approach - Blades' automatic skip propagation for complex routing scenarios
Source Files