func (b *Builder) Validate() error {
	stepCnt := len(b.steps)
	if stepCnt < 2 {
		return errStepCnt
	}
	var (
		input      reflect.Type
		output     reflect.Type
		prevOutput reflect.Type
		st         stepType
	)
	for i := range b.steps {
		st = typeOfStep(b.steps[i].step)
		if st == invalid {
			return errStepType
		} else if i == 0 && st == onlyIn {
			return errFirstStep
		} else if i == stepCnt-1 && st == onlyOut {
			return errLastStep
		} else if 0 < i && i < stepCnt-1 && st != inOut {
			return errInteriorStep
		}
		b.steps[i].sType = st
		switch st {
		case onlyOut:
			output = reflect.TypeOf(b.steps[i].step).Out(0)
		case inOut:
			input = reflect.TypeOf(b.steps[i].step).In(1)
			output = reflect.TypeOf(b.steps[i].step).Out(0)
		case onlyIn:
			input = reflect.TypeOf(b.steps[i].step).In(1)
		}
		if i == 0 {
			prevOutput = output
			continue
		}
		if input.Kind() == reflect.Interface {
			if !prevOutput.Implements(input) {
				return fmt.Errorf(typeMismatchFmt, i+1, prevOutput, input)
			}
		} else if prevOutput != input {
			return fmt.Errorf(typeMismatchFmt, i+1, prevOutput, input)
		}
		prevOutput = output
	}
	if b.inCh != nil {
		err := validateInputChannel(b.inCh, b.steps[0])
		if err != nil {
			return err
		}
	}
	if b.outCh != nil {
		err := validateOutputChannel(b.outCh, b.steps[len(b.steps)-1])
		if err != nil {
			return err
		}
	}
	return nil
}
Validate makes sure the pipeline can process data. It ensures all registered steps are of the right type and that
their input and output types line up. This is the methodd that BuildAndExecute calls. It is exposed mainly for
testing purposes so that end users of this api can find out at compile time if their pipeline is set up correctly.
package flo
import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sync"
)
var (
	errStepCnt              = errors.New("must register at least two steps")
	errFirstStep            = errors.New("first step must have a signature of func(context.Context) (R, error) or func(context.Context, T) (R, error)")
	errInteriorStep         = errors.New("interior step must have a signature of func(context.Context, T) (R, error)")
	errLastStep             = errors.New("last step must have a signature of func(context.Context, T) error")
	errStepType             = errors.New("a Step must be func with one of the following signatures: func(context.Context) (R, error), func(context.Context, T) (R, error), or func(context.Context, T) error")
	errInputChType          = errors.New("a input channel must be of type <-chan T")
	errInputChStepType      = errors.New("a input channel should only be registered when first step is of type func(context.Context, T) (R, error)")
	errOutputChType         = errors.New("an output channel must be of type chan<- T")
	errOutputChStepType     = errors.New("an output channel should only be registered when last step is of type func(context.Context, T) (R, error)")
	inputChTypeMismatchFmt  = "input channels type %s does not match the first steps input type %s"
	outputChTypeMismatchFmt = "output channels type %s does not match the last steps output type %s"
	typeMismatchFmt         = "Step %d: previous steps output type %s does not match current steps input type %s"
)
type Builder struct {
	inCh        interface{}
	outCh       interface{}
	realChan    chan interface{}
	steps       []*stepRunner
	parallelism int
	errHandler  func(error)
}
type Option func(*Builder)
func WithParallelism(i int) Option {
	return func(b *Builder) {
		if i < 1 {
			i = 1
		}
		b.parallelism = i
	}
}
func WithErrorHandler(handler ErrorHandler) Option {
	return func(b *Builder) {
		b.errHandler = handler
	}
}
func WithInput(ch interface{}) Option {
	return func(b *Builder) {
		b.inCh = ch
	}
}
func WithOutput(ch interface{}) Option {
	return func(b *Builder) {
		b.outCh = ch
	}
}
func NewBuilder(options ...Option) *Builder {
	f := &Builder{
		parallelism: 1,
	}
	for i := range options {
		options[i](f)
	}
	return f
}
func (b *Builder) Add(s Step, options ...StepOption) *Builder {
	sr := &stepRunner{
		step:        s,
		parallelism: b.parallelism,
		wg:          &sync.WaitGroup{},
		errHandler:  b.errHandler,
	}
	for i := range options {
		options[i](sr)
	}
	b.steps = append(b.steps, sr)
	return b
}
func (b *Builder) BuildAndExecute(ctx context.Context) error {
	err := b.Validate()
	if err != nil {
		return err
	}
	for i := range b.steps {
		if i == 0 {
			if b.inCh != nil {
				b.steps[i].registerInput(b.launchInputChannel())
			}
		} else {
			b.steps[i].registerInput(b.steps[i-1].output())
		}
		if b.steps[i].sType != onlyIn {
			b.steps[i].output()
		}
		b.steps[i].start(ctx)
	}
	if b.outCh != nil {
		b.launchOutputChannel()
	}
	b.awaitShutdown()
	return nil
}