chain

package module
v1.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 20, 2026 License: MIT Imports: 7 Imported by: 0

README

Chain

chain is a small Go workflow library built around composable Action values. An action receives one value, returns the next value, and can be connected to other actions through a Workflow.

The package focuses on in-process workflow control. It does not try to be a distributed scheduler or a full parallel DAG engine.

Installation

go get github.com/JSYoo5B/chain

Core Concepts

Action

An Action is the basic execution unit.

type Action[T any] interface {
    Name() string
    Run(ctx context.Context, input T) (output T, err error)
}

Use NewSimpleAction when a function is enough:

double := chain.NewSimpleAction(
    "double",
    func(ctx context.Context, input int) (int, error) {
        return input * 2, nil
    },
)
Workflow

A Workflow connects actions and also implements Action, so workflows can be nested inside other workflows.

workflow := chain.NewWorkflow("calculation", action1, action2, action3)

output, err := workflow.Run(context.Background(), input)

By default, actions are connected through the Success direction in constructor order. Failure and Abort terminate unless a custom RunPlan is set.

workflow.SetRunPlan(action1, chain.DefaultPlan(action2, errorHandler))
Directions and Run Plans

A RunPlan maps a direction to the next action.

type RunPlan[T any] map[string]Action[T]

Built-in directions are:

  • Success
  • Failure
  • Abort

BranchAction can add custom directions.

Action Builders

Basic Actions
  • NewSimpleAction creates an action from a function.
  • NewSimpleBranchAction creates a branch action from functions.
  • AsBranchAction wraps an existing action and adds branching logic.
Error Control
  • AsRetryableAction retries a main action and optionally runs rollback before the next attempt.
  • SkipRollback explicitly disables rollback for a retryable action.
  • AsBestEffortAction suppresses a non-critical action error and optionally calls a fallback hook.
Collection Processing
  • AsSequenceSliceAction runs an action over a slice sequentially.
  • AsSequenceMapAction runs an action over a map sequentially.
  • AsParallelSliceAction runs an action over a slice concurrently.
  • AsParallelMapAction runs an action over a map concurrently.

Parallel collection actions preserve output positions or keys, but joined error order follows completion order.

Type Adaptation
  • AdaptAction runs an action against a field or sub-value inside a larger value.

This is useful when a workflow carries one aggregate state type, while some actions only operate on one part of that state.

Validation

Workflow.ValidateGraph checks the configured workflow graph before execution.

It rejects:

  • directed cycles
  • disconnected workflow graphs

Branching and merge points are allowed as long as the graph remains a connected DAG.

Examples

See:

Documentation

Overview

Package chain provides small, composable workflow primitives.

The basic unit is Action. A Workflow connects actions with RunPlan values and moves between them through directions such as Success, Failure, Abort, or custom directions returned by BranchAction implementations.

The package also includes wrappers for common execution patterns:

  • AsBranchAction and NewSimpleBranchAction for direction-based branching.
  • AsRetryableAction for bounded retries and optional rollback.
  • AsBestEffortAction for non-critical work that should not fail a workflow.
  • AsSequenceSliceAction and AsSequenceMapAction for sequential collection processing.
  • AsParallelSliceAction and AsParallelMapAction for parallel collection processing.
  • AdaptAction for running an action against a field or sub-value inside a larger value.

Index

Constants

View Source 
const (
	// Success represents the direction indicating that the action completed successfully
	// and the Workflow should continue.
	Success = "success"

	// Failure represents the direction indicating that an error occurred,
	// and the Workflow should handle it accordingly.
	Failure = "failure"

	// Abort represents the direction indicating that
	// the Workflow execution should be aborted immediately.
	// This can occur due to a specific Abort condition or
	// in cases of unexpected errors or panics that cause the Workflow to halt.
	Abort = "abort"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type Action 

type Action[T any] interface {
	// Name provides the identifier of this Action.
	Name() string

	// Run executes the Action, processing the input and returning output or an error.
	Run(ctx context.Context, input T) (output T, err error)
}

Action is the basic unit of execution in this package. It represents a single task that processes input and produces output.

func AdaptAction added in v1.2.0 

func AdaptAction[T any, U any](
	action Action[U],
	getter InternalTypeGetter[T, U],
	setter ExternalTypeSetter[T, U],
) Action[T]

AdaptAction creates an Action that works with a composite data structure (T), where T is a complex type (e.g., a struct with multiple fields) and U is the data type that the Action operates on. The InternalTypeGetter and ExternalTypeSetter functions extract U from T and write the processed result back into T.

func AsBestEffortAction added in v1.3.0 

func AsBestEffortAction[T any](action Action[T], fallback BestEffortFallbackFunc[T]) Action[T]

AsBestEffortAction creates an Action that suppresses errors from the wrapped action.

The wrapped action keeps its original Name behavior. When the wrapped action succeeds, its output is returned normally. When it returns an error, fallback is called with the original input and error, then the wrapped action's output is returned with a nil error. This allows a Workflow to continue through the Success direction even when the wrapped action failed in a non-critical way.

func AsParallelMapAction added in v1.2.0 

func AsParallelMapAction[K comparable, T any](name string, action Action[T]) Action[map[K]T]

AsParallelMapAction creates an Action that processes a map's values in parallel. Each value is transformed concurrently while output keys are preserved.

The action handles panics gracefully, continuing execution of other goroutines when one fails. If any error or panic occurs, the action returns an error but still provides the processed output for successful operations. Errors from multiple keys are joined in completion order.

func AsParallelSliceAction added in v1.2.0 

func AsParallelSliceAction[T any](name string, action Action[T]) Action[[]T]

AsParallelSliceAction creates an Action that processes a slice's elements in parallel. Each element is transformed concurrently while output order is preserved.

The action handles panics gracefully, continuing execution of other goroutines when one fails. If any error or panic occurs, the action returns an error but still provides the processed output for successful operations. Errors from multiple elements are joined in completion order.

func AsRetryableAction added in v1.2.0 

func AsRetryableAction[T any](name string, mainAction, rollbackAction Action[T], maxRetry int) Action[T]

AsRetryableAction creates an Action that retries the mainAction up to maxRetry times.

If the mainAction fails, it executes the rollbackAction before the next retry attempt. The rollbackAction is not executed on the final attempt if it fails. rollbackAction can be skipped when it is nil.

func AsSequenceMapAction added in v1.2.0 

func AsSequenceMapAction[K comparable, T any](name string, action Action[T]) Action[map[K]T]

AsSequenceMapAction creates an Action that processes a map's values sequentially. Each value is transformed one at a time while output keys are preserved.

Errors from multiple keys are joined. Map iteration order is not deterministic, so callers should not depend on the order of joined errors.

Unlike parallel processing, sequential execution stops immediately when a panic occurs, leaving unprocessed values unchanged in the output.

func AsSequenceSliceAction added in v1.2.0 

func AsSequenceSliceAction[T any](name string, action Action[T], stopOnError bool) Action[[]T]

AsSequenceSliceAction creates an Action that processes a slice's elements sequentially. Each element is transformed one at a time while output order is preserved.

The stopOnError parameter controls error handling behavior:

  • When true, processing stops on the first error.
  • When false, processing continues and errors are joined.

Panics always stop execution regardless of the stopOnError setting.

func NewSimpleAction 

func NewSimpleAction[T any](name string, runFunc RunFunc[T]) Action[T]

NewSimpleAction creates a new Action with a custom Run function, which can be a pure function or closure. The provided runFunc must match the RunFunc signature.

This allows simple Actions to be created without manually defining a struct.

func SkipRollback added in v1.2.1 

func SkipRollback[T any]() Action[T]

SkipRollback provides an Action that explicitly skips the rollback process for a RetryableAction.

When creating a RetryableAction with AsRetryableAction, use SkipRollback() as the rollback action to make the intent clear, rather than passing raw nil.

func Terminate 

func Terminate[T any]() Action[T]

Terminate provides an Action that explicitly stops the execution of a Workflow.

When returned from a RunPlan, it signals that the Workflow should halt and no further actions should be executed. This is clearer than returning raw nil.

type BestEffortFallbackFunc added in v1.3.0 

type BestEffortFallbackFunc[T any] func(ctx context.Context, input T, err error)

BestEffortFallbackFunc is called when a wrapped action returns an error. The input argument is the value that was passed to the wrapped action.

type BranchAction 

type BranchAction[T any] interface {
	// Name returns the name of the BranchAction.
	Name() string

	// Run executes the branch action.
	Run(ctx context.Context, input T) (output T, err error)

	// Directions returns custom directions that NextDirection can return.
	// Built-in directions are already available in Workflow run plans.
	Directions() []string

	// NextDirection selects the next execution path from the Run output.
	// It is called only if Run succeeds (err == nil).
	NextDirection(ctx context.Context, output T) (direction string, err error)
}

BranchAction controls workflow branching after its Run method succeeds. It extends Action with custom directions and a direction selector.

func AsBranchAction added in v1.3.0 

func AsBranchAction[T any](baseAction Action[T], branchFunc BranchFunc[T], directions ...string) BranchAction[T]

AsBranchAction extends an existing Action with branching logic. The wrapped action keeps its original Name and Run behavior, while branchFunc determines the next direction from the wrapped action's output.

The optional directions argument describes custom directions returned by branchFunc. Built-in directions such as Success, Failure, and Abort are already available in Workflow plans.

func NewSimpleBranchAction 

func NewSimpleBranchAction[T any](name string, runFunc RunFunc[T], directions []string, branchFunc BranchFunc[T]) BranchAction[T]

NewSimpleBranchAction creates a new BranchAction with customizable directions. It accepts a name, custom directions, and a BranchFunc that selects the next direction from the action's output.

A custom runFunc can be provided to define execution logic. If runFunc is nil, the action passes the input through as output.

This allows simple BranchActions to be created without manually defining a struct that implements BranchAction.

type BranchFunc 

type BranchFunc[T any] func(ctx context.Context, output T) (direction string, err error)

BranchFunc represents the signature for the function that defines the branching logic for a BranchAction in the package. It takes the running context and output as input and returns the direction for the next step along with any error.

type ExternalTypeSetter added in v1.2.0 

type ExternalTypeSetter[T any, U any] func(T, U) T

ExternalTypeSetter updates a composite data structure (T) with a new subpart (U). It returns the updated composite value.

type InternalTypeGetter added in v1.2.0 

type InternalTypeGetter[T any, U any] func(T) U

InternalTypeGetter extracts a subpart (U) from a composite data structure (T). It allows access to a part of the structure without modifying the entire one.

type RunFunc 

type RunFunc[T any] func(ctx context.Context, input T) (output T, err error)

RunFunc is the function signature used by NewSimpleAction. It receives an input value and returns the next value or an error.

type RunPlan added in v1.2.2 

type RunPlan[T any] map[string]Action[T]

RunPlan maps directions to the next Action to execute. Directions can be built-in values such as Success, Failure, and Abort, or custom directions returned by a BranchAction.

func DefaultPlan 

func DefaultPlan[T any](success, error Action[T]) RunPlan[T]

DefaultPlan returns a standard RunPlan with valid next actions for Success and Failure, and Termination for Abort.

func DefaultPlanWithAbort 

func DefaultPlanWithAbort[T any](success, error, abort Action[T]) RunPlan[T]

DefaultPlanWithAbort returns a RunPlan for Success, Failure, and Abort.

func SuccessOnlyPlan 

func SuccessOnlyPlan[T any](success Action[T]) RunPlan[T]

SuccessOnlyPlan returns a RunPlan where only Success has a next action. Failure and Abort both lead to termination.

func TerminationPlan 

func TerminationPlan[T any]() RunPlan[T]

TerminationPlan returns a RunPlan that terminates immediately.

type Workflow added in v1.1.1 

type Workflow[T any] struct {
	// contains filtered or unexported fields
}

Workflow represents Actions connected by direction-based run plans. It executes each of its constituent Actions in sequence, with each Action following its own Run method. The flow proceeds based on the defined structure of the Workflow, allowing flexible and organized execution of actions to build workflows that can be as simple or complex as needed.

Workflow implements Action, so it can be nested inside another Workflow.

func NewWorkflow added in v1.1.1 

func NewWorkflow[T any](name string, memberActions ...Action[T]) *Workflow[T]

NewWorkflow creates a new Workflow by taking a series of Actions as its members. These Actions are connected through Success in the order they are provided. Custom run plans can replace the default flow.

func (*Workflow[T]) Name added in v1.1.1 

func (w *Workflow[T]) Name() string

Name provides the identifier of this Workflow.

func (*Workflow[T]) Run added in v1.1.1 

func (w *Workflow[T]) Run(ctx context.Context, input T) (output T, err error)

Run executes the Workflow by running Actions in the order they were configured, starting from the initAction, which is the first one of the memberActions provided by the constructor such as NewWorkflow. Each action receives the previous action's output as input.

func (*Workflow[T]) RunAt added in v1.1.1 

func (w *Workflow[T]) RunAt(initAction Action[T], ctx context.Context, input T) (output T, err error)

RunAt starts the execution of the Workflow from a given Action (initAction). It follows run plans and moves between actions by direction. If an action returns an error, the Workflow proceeds through Failure. Abort immediately halts execution unless the plan specifies otherwise. If no action plan is found for a given direction, the Workflow will terminate with the appropriate error.

func (*Workflow[T]) SetRunPlan added in v1.1.1 

func (w *Workflow[T]) SetRunPlan(currentAction Action[T], plan RunPlan[T])

SetRunPlan updates the execution flow for the given currentAction in the Workflow by associating it with a specified RunPlan. The currentAction will be validated to ensure it is a member of the Workflow. The RunPlan defines the directions (such as Success, Failure, Abort) and their corresponding next actions.

If the currentAction is nil or not part of the Workflow, a panic will occur. The plan can be nil, in which case the currentAction will be set to terminate for any direction not explicitly specified in the plan. If a direction is encountered in the plan that is not valid for the currentAction, or if it leads to an invalid action, another panic will occur.

Additionally, self-loops are not allowed in the plan. If the next action for a direction is the current action itself, a panic will be triggered.

func (*Workflow[T]) ValidateGraph added in v1.1.1 

func (w *Workflow[T]) ValidateGraph() error

ValidateGraph ensures the workflow's graph is connected and acyclic. It checks for cycles first, then verifies that all nodes are connected.

Source Files

View all Source files

Directories

Path Synopsis
internal
errors
logger

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.