chain

README

go-chain

A Chain of Responsibility for Go, built on channels: a value passes through a chain of steps, each step does one thing to it — turns it into another, routes it, drops it — and hands it on. Every step runs in its own goroutine; steps are connected by typed channels; a chain may branch, join, run a step on several workers and nest other chains. Generic, standard library only.

go get github.com/eggs-gd/go-chain

in, lengths := make(chan string), make(chan int)
even, odd := make(chan int), make(chan int)

c := chain.NewChainProcessor(errs)                                     // every step's errors
c.AddStep(chain.NewEntryPoint(in, words(text)))                        // the one input
c.AddStep(chain.NewDecoratorN(in, lengths, length{}, 4))               // one step on 4 workers
c.AddStep(chain.NewSwitch(lengths, []chan<- int{even, odd}, parity{})) // a branch
c.AddStep(chain.NewEnd(even, collect{&evens}))
c.AddStep(chain.NewEnd(odd, collect{&odds}))
c.Process(ctx)                                                         // one pass, to its end

The whole example: example_test.go.

The pattern

Chain of Responsibility splits processing into handlers, each knowing one responsibility and nothing of the others; a request travels the chain, each handler deciding what to do with it. Here a handler is a step, and the chain is a pipeline: many values travel at once, each step working on its own value concurrently with the others. A step's logic is plain Go — an interface of one method; the package runs it:

logic	method	as a step
EntryPoint[T]	Start(ctx, emit func(T) bool) error	the chain's one input: emits its values, then is done
Decorator[Ti, To]	Decorate(Ti) (To, error)	one value in, one out
Switcher[T]	Switch(T) (int, error)	a value to one of several outputs (its index)
SwitchDecorator[Ti, To]	Switch(Ti) (int, To, error)	a value, turned into another, to one output
Consumer[T]	Consume(T) error	a chain's end: every value consumed

Constructors connect a logic to channels: NewEntryPoint(out, e), NewDecorator(in, out, d), NewSwitch(in, outs, s), NewSwitchDecorator(in, outs, s), NewEnd(in, c), and NewChainProcessor(errs) + AddStep for a chain. Optional on any logic:

• Flusher[To] (Flush() ([]To, error)): what the step holds goes out when its input ends — a group not complete yet, a batch not full.
• Stopper (Stop()): called once, when the step ends — to close what it opened.

A pass

Process is one pass. A chain has one input, its entry point: it emits its values and returns, and its output closes. Every other step reads its input until it closes, gives what it holds (Flusher) and returns. An output closes once every step that writes to it has returned — the chain keeps a sync.WaitGroup per output, one count per writer. So the end of the input travels the chain step by step, and Process returns when every step has: by then every value of the pass went through every step. No end-of-stream value in the data, no counting.

Channels close once: the next pass is a new chain with new channels.

Branches and joins

               ┌─▶ even ─▶ [end]
[entry] ─▶ [switch]
               └─▶ odd ─▶ [decorator] ─┐
                                       ├─▶ joint ─▶ [end]
                           [another] ──┘

• A switch sends each value to one of its outputs; when its input ends, all its outputs close. A negative index drops the value on purpose; an index past the outputs is an error, not a value lost silently.
• A join is a channel several steps write to: it closes after the last of them returned, however long a branch takes.

Workers

NewDecoratorN, NewSwitchN, NewSwitchDecoratorN run the same logic on n goroutines reading one input (n last, as in strings.SplitN). The logic must be safe for concurrent use; the order of values may change. The step returns after its last worker: its output closes only when every value is done.

Sub-chains

A chain is a step too: AddStep(subchain). When it runs inside another chain, its steps join the outer chain's: the writers of every channel are counted across all levels, so a channel written by steps of two sub-chains closes after the last of them. Made with NewChainProcessor(nil), a sub-chain reports its errors to the chain it runs in — a stage of a bigger pipeline declares its own steps in its own constructor, and the top stays a list of stages.

Errors and skips

• A step's error goes to the chain's error channel; the value is dropped, the pass goes on.
• ErrSkippedItem — a value dropped on purpose (buffered, unchanged, not wanted) — is not an error: it never reaches the error channel.
• An entry point's Start error is reported the same way.

Cancellation

Every send — to a channel or to the error channel — gives up when the context ends, so a cancelled pass never blocks on a full channel; emit returns false and the entry point should return. A step stopped by the context does not flush.

Tests

go test -race ./...: a pass, a flusher's values before its output closes, a switch and a join that closes after its slow branch, a join across two sub-chains, a switch index out of range, workers that finish every value, errors and skips, a sub-chain inheriting the error channel, cancel unblocking a send, Stop called once; the example.

License

MIT

Documentation

Overview

Package chain: a Chain of Responsibility on channels — steps that run concurrently, connected by typed channels. A chain has one input, its entry point; a pass (Process) runs every step until its input closes: the entry closes its output when it is done, a step that read its input to the end gives what it holds (Flusher) and returns, and an output closes once every step that writes to it has returned (a WaitGroup per output: where branches join, the joint closes after the last of them). Process returns when every step has.

Example

A pass: words → lengths (on 4 workers) → split by parity → two ends. Process returns once every value went through every step.

package main

import (
	"context"
	"fmt"
	"slices"
	"strings"

	chain "github.com/eggs-gd/go-chain"
)

// words: the chain's entry — every word of a text, then it is done
type words string

func (w words) Start(ctx context.Context, emit func(string) bool) error {
	for _, word := range strings.Fields(string(w)) {
		if !emit(word) {
			return ctx.Err()
		}
	}
	return nil
}

// length: a decorator — a word into its length (empty ones skipped, not an error)
type length struct{}

func (length) Decorate(w string) (int, error) {
	if w == "-" {
		return 0, chain.ErrSkippedItem
	}
	return len(w), nil
}

// parity: a switch — even lengths to output 0, odd ones to output 1
type parity struct{}

func (parity) Switch(n int) (int, error) { return n % 2, nil }

// collect: a chain's end
type collect struct{ got *[]int }

func (c collect) Consume(n int) error { *c.got = append(*c.got, n); return nil }

// A pass: words → lengths (on 4 workers) → split by parity → two ends. Process
// returns once every value went through every step.
func main() {
	in, lengths := make(chan string), make(chan int)
	even, odd := make(chan int), make(chan int)
	var evens, odds []int

	c := chain.NewChainProcessor(nil)
	c.AddStep(chain.NewEntryPoint(in, words("a chain of - small steps")))
	c.AddStep(chain.NewDecoratorN(in, lengths, length{}, 4))
	c.AddStep(chain.NewSwitch(lengths, []chan<- int{even, odd}, parity{}))
	c.AddStep(chain.NewEnd(even, collect{&evens}))
	c.AddStep(chain.NewEnd(odd, collect{&odds}))
	c.Process(context.Background())

	slices.Sort(evens) // 4 workers: the order may change
	slices.Sort(odds)
	fmt.Println(evens, odds)
}

Output:
[2] [1 5 5 5]

Index

• Variables
• type ChainProcessor
  • func NewChainProcessor(errch chan<- error) ChainProcessor
• type Consumer
• type Decorator
• type EntryPoint
• type Flusher
• type Processor
  • func NewDecorator[Ti, To any](in <-chan Ti, out chan<- To, logic Decorator[Ti, To]) Processor
  • func NewDecoratorN[Ti, To any](in <-chan Ti, out chan<- To, logic Decorator[Ti, To], n int) Processor
  • func NewEnd[T any](in <-chan T, logic Consumer[T]) Processor
  • func NewEntryPoint[T any](out chan<- T, logic EntryPoint[T]) Processor
  • func NewSwitch[T any](in <-chan T, outs []chan<- T, logic Switcher[T]) Processor
  • func NewSwitchDecorator[Ti, To any](in <-chan Ti, outs []chan<- To, logic SwitchDecorator[Ti, To]) Processor
  • func NewSwitchDecoratorN[Ti, To any](in <-chan Ti, outs []chan<- To, logic SwitchDecorator[Ti, To], n int) Processor
  • func NewSwitchN[T any](in <-chan T, outs []chan<- T, logic Switcher[T], n int) Processor
• type Stopper
• type SwitchDecorator
• type Switcher

Examples

• Package

Variables

var ErrSkippedItem = errors.New("skipped item")

ErrSkippedItem: a step drops the value on purpose — not an error, never reported

Types

type ChainProcessor

type ChainProcessor interface {
	Processor
	AddStep(p Processor)
	Process(ctx context.Context)
}

ChainProcessor: steps that run together, one pass (Process); a chain is a step too (a sub-chain)

func NewChainProcessor

func NewChainProcessor(errch chan<- error) ChainProcessor

NewChainProcessor: errch gets the steps' errors (nil: the errors of the chain this one runs in)

type Consumer

type Consumer[T any] interface {
	Consume(T) error
}

Consumer: the logic of a chain's end — every value consumed. Optional: Stopper.

type Decorator

type Decorator[Ti, To any] interface {
	Decorate(Ti) (To, error)
}

Decorator: a step's logic — one value in, one out (or an error). Optional: Stopper, Flusher.

type EntryPoint

type EntryPoint[T any] interface {
	Start(ctx context.Context, emit func(T) bool) error
}

EntryPoint: the logic of a chain's entry — one pass: every value it finds, emitted (emit is false once the chain stops). Optional: Stopper.

type Flusher

type Flusher[To any] interface {
	Flush() ([]To, error)
}

Flusher: a step's logic may hold values; when its input ends they go out

type Processor

type Processor interface {
	// contains filtered or unexported methods
}

Processor: a step, or a chain of steps

func NewDecorator

func NewDecorator[Ti, To any](in <-chan Ti, out chan<- To, logic Decorator[Ti, To]) Processor

NewDecorator: every value of in through the logic, to out

func NewDecoratorN

func NewDecoratorN[Ti, To any](in <-chan Ti, out chan<- To, logic Decorator[Ti, To], n int) Processor

NewDecoratorN: the same on n workers (the logic safe for concurrent use; the order may change)

func NewEnd

func NewEnd[T any](in <-chan T, logic Consumer[T]) Processor

NewEnd: a chain's end

func NewEntryPoint

func NewEntryPoint[T any](out chan<- T, logic EntryPoint[T]) Processor

NewEntryPoint: the chain's one input, an output only; when it is done, its output closes

func NewSwitch

func NewSwitch[T any](in <-chan T, outs []chan<- T, logic Switcher[T]) Processor

NewSwitch: a value to one output; when the input ends, every output is done

func NewSwitchDecorator

func NewSwitchDecorator[Ti, To any](in <-chan Ti, outs []chan<- To, logic SwitchDecorator[Ti, To]) Processor

NewSwitchDecorator: a value, turned into another, to one output

func NewSwitchDecoratorN

func NewSwitchDecoratorN[Ti, To any](in <-chan Ti, outs []chan<- To, logic SwitchDecorator[Ti, To], n int) Processor

NewSwitchDecoratorN: the same on n workers (the order may change)

func NewSwitchN

func NewSwitchN[T any](in <-chan T, outs []chan<- T, logic Switcher[T], n int) Processor

NewSwitchN: the same on n workers (the order may change)

type Stopper

type Stopper interface{ Stop() }

Stopper: a step's logic may have one; it is called once, when the step ends

type SwitchDecorator

type SwitchDecorator[Ti, To any] interface {
	Switch(Ti) (int, To, error)
}

SwitchDecorator: a switch's logic that also turns the value into another (the output's index and the new value). Optional: Stopper.

type Switcher

type Switcher[T any] interface {
	Switch(T) (int, error)
}

Switcher: a step's logic that picks one output for a value (its index; negative: none, the value dropped on purpose; past the outputs: an error). Optional: Stopper.