gorunner

README

gorunner

gorunner is a small Go library for managing concurrent executions of typed pipelines (called InstructionSets) with retry support and type safety checks at runtime. You can define a series of functions (Instructions) that transform data of a specific type. A Runner orchestrates these sets, enforcing concurrency limits and automatically retrying failed executions.

Features

• Typed pipelines: Each InstructionSet is bound to a concrete type, preventing accidental type mixing.
• Concurrent execution: Control concurrency through a semaphore mechanism.
• Retry support: Automatically re-run a pipeline upon failure, up to a specified limit.
• Callbacks: Optional onErr and onSuccess callbacks to handle outcomes.

Disclaimer - ChatGPT was used to write documentation and tests.

Documentation

Index

• type Instruction
• type InstructionSet
  • func NewInstructionSet[T any](name string, onErr func(error, ...T), onSuccess func(T)) *InstructionSet[T]
  • func (i *InstructionSet[T]) Add(instruction Instruction[T]) *InstructionSet[T]
  • func (i *InstructionSet[T]) IncludeDataWithError() *InstructionSet[T]
  • func (i *InstructionSet[T]) Save(runner *Runner) error
• type Runner
  • func New(retries, concurrency int) *Runner
  • func (r *Runner) RemoveSet(name string)
  • func (r *Runner) Run(name string, args any) error

Types

type Instruction

type Instruction[T any] func(shared T) (T, error)

Instruction represents a single function in an InstructionSet pipeline. It receives a value of type T and returns a transformed value of type T and an error if something goes wrong.

type InstructionSet

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

InstructionSet holds a series of Instruction[T], an optional onErr handler, an optional onSuccess handler, and a flag to indicate whether the original input data should be provided to onErr. It uses a builder-like pattern for configuration.

func NewInstructionSet

func NewInstructionSet[T any](
	name string,
	onErr func(error, ...T),
	onSuccess func(T),
) *InstructionSet[T]

NewInstructionSet creates a new typed InstructionSet, identified by a unique name.

The onErr callback is optional. When invoked, it will receive the final error, and if provideToErr is set, also the original input data.

The onSuccess callback is optional. It will be invoked with the final successfully processed data after all Instructions complete without error.

Example usage:

runner := gorunner.New(10, 10)
set := gorunner.NewInstructionSet[int]("count",
    func(err error, data ...int) { fmt.Println("Error:", err, data) },
    func(result int) { fmt.Println("Success:", result) },
)

set.Add(func(data int) (int, error) { return data + 1, nil }).
    Add(func(data int) (int, error) { return data + 3, nil }).
    IncludeDataWithError().
    Save(runner)

err := runner.Run("count", 42)

func (*InstructionSet[T]) Add

func (i *InstructionSet[T]) Add(instruction Instruction[T]) *InstructionSet[T]

Add appends a new Instruction[T] to the InstructionSet. Instructions will be executed in the order they are added when the InstructionSet is invoked.

func (*InstructionSet[T]) IncludeDataWithError

func (i *InstructionSet[T]) IncludeDataWithError() *InstructionSet[T]

IncludeDataWithError is an optional configuration for an InstructionSet. When set, it ensures that the onErr callback receives the original arguments provided to Run alongside the error.

func (*InstructionSet[T]) Save

func (i *InstructionSet[T]) Save(runner *Runner) error

Save registers the current InstructionSet with the given Runner. If an InstructionSet with the same name already exists in the Runner, an error is returned. Otherwise, it is stored and can be invoked by calling runner.Run(i.name, ...).

type Runner

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

Runner manages named InstructionSet(s) and invokes them concurrently up to a specified limit.

A Runner is configured with a retry count and a maximum concurrency level. Each named InstructionSet can be invoked by calling Run. Type-checking is performed at runtime to ensure that the provided arguments match the expected type of the InstructionSet.

func New

func New(retries, concurrency int) *Runner

New creates and returns a new Runner.

The 'retries' parameter specifies how many times an InstructionSet will be retried upon error. The 'concurrency' parameter specifies the maximum number of concurrent InstructionSet invocations allowed.

Example usage:

runner := gorunner.New(10, 10)
set := gorunner.NewInstructionSet[int]("count", nil, nil)
set.Save(runner)
err := runner.Run("count", 42)

func (*Runner) RemoveSet

func (r *Runner) RemoveSet(name string)

RemoveSet removes the InstructionSet associated with the given name from the Runner.

If no InstructionSet is found for the provided name, this function does nothing.

func (*Runner) Run

func (r *Runner) Run(name string, args any) error

Run invokes the named InstructionSet with the provided arguments. If the InstructionSet does not exist, an error is returned immediately. If the argument type does not match what the InstructionSet expects, an error is also returned immediately.

Run returns an error only if the InstructionSet is not found or if the argument type is invalid. It does not wait for the instructions to finish. If the InstructionSet exists and the type check passes, Run starts the execution asynchronously and returns nil.

Concurrency is limited by the Runner’s internal semaphore. If concurrency has reached its limit, the call to Run will block until a "slot" is available.