boost

type AnnotatedWaitGroup ¶

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

AnnotatedWaitGroup is a wrapper around the standard WaitGroup that provides annotations to wait group operations that can assist in diagnosing concurrency issues.

func (d *AnnotatedWaitGroup) Add(delta int, name ...GoRoutineName)

func (d *AnnotatedWaitGroup) Count() int

func (d *AnnotatedWaitGroup) Done(name ...GoRoutineName)

func (d *AnnotatedWaitGroup) Wait(name ...GoRoutineName)

type AnnotatedWgAQ ¶

type AnnotatedWgAQ interface {
	AnnotatedWgAdder
	AnnotatedWgQuitter
}

AnnotatedWgAQ is the interface that is a restricted view of a wait group that allows adding to the wait group and Done signalling with the addition of being able to specify the name representing the calling go routine. This interface can be acquired from the wait group using a standard interface type query.

type AnnotatedWgAdder ¶

type AnnotatedWgAdder interface {
	Add(delta int, name ...GoRoutineName)
}

AnnotatedWgAdder is the interface that is a restricted view of a wait group that only allows adding to the wait group with the addition of being able to specify the name representing the calling go routine. This interface can be acquired from the wait group using a standard interface type query.

type AnnotatedWgCounter ¶

type AnnotatedWgCounter interface {
	Count() int
}

AnnotatedWgCounter is the interface that is a restricted view of a wait group that only allows querying the wait group count. This interface can be acquired from the wait group using a standard interface type query.

type AnnotatedWgQuitter ¶

type AnnotatedWgQuitter interface {
	Done(name ...GoRoutineName)
}

AnnotatedWgQuitter is the interface that is a restricted view of a wait group that only allows Done signalling on the wait group with the addition of being able to specify the name representing the calling go routine. This interface can be acquired from the wait group using a standard interface type query.

type AnnotatedWgWaiter ¶

type AnnotatedWgWaiter interface {
	Wait(name ...GoRoutineName)
}

AnnotatedWgWaiter is the interface that is a restricted view of a wait group that only allows waiting on the wait group with the addition of being able to specify the name representing the calling go routine. This interface can be acquired from the wait group using a standard interface type query.

type CancelStream ¶

type CancelStream = chan CancelWorkSignal

type CancelStreamR ¶

type CancelStreamR = <-chan CancelWorkSignal

type CancelStreamW ¶

type CancelStreamW = chan<- CancelWorkSignal

type CancelWorkSignal ¶

type CancelWorkSignal struct{}

type Duplex ¶ added in v0.4.0

type Duplex[T any] struct {
	Channel  chan T
	ReaderCh <-chan T
	WriterCh chan<- T
}

Duplex represents a channel with multiple views, to be used by clients that need to hand out different ends of the same channel to different entities.

func NewDuplex[T any](channel chan T) *Duplex[T]

type DuplexJobOutput ¶ added in v0.4.0

type DuplexJobOutput[O any] Duplex[JobOutput[O]]

type ExecutiveFunc ¶

type ExecutiveFunc[I, O any] func(j Job[I]) (JobOutput[O], error)

func (f ExecutiveFunc[I, O]) Invoke(j Job[I]) (JobOutput[O], error)

type GoRoutineID ¶

type GoRoutineID string

type GoRoutineName ¶

type GoRoutineName string

type Job ¶

type Job[I any] struct {
	ID         string
	Input      I
	SequenceNo int
}

type JobOutput ¶

type JobOutput[O any] struct {
	Payload O
}

type JobOutputStream ¶ added in v0.4.0

type JobOutputStream[O any] chan JobOutput[O]

type JobOutputStreamR ¶ added in v0.4.0

type JobOutputStreamR[O any] <-chan JobOutput[O]

type JobOutputStreamW ¶ added in v0.4.0

type JobOutputStreamW[O any] chan<- JobOutput[O]

type JobStream ¶

type JobStream[I any] chan Job[I]

type JobStreamR ¶

type JobStreamR[I any] <-chan Job[I]

type JobStreamW ¶

type JobStreamW[I any] chan<- Job[I]

type NewWorkerPoolParams ¶

type NewWorkerPoolParams[I, O any] struct {
	NoWorkers       int
	OutputChTimeout time.Duration
	Exec            ExecutiveFunc[I, O]
	JobsCh          JobStream[I]
	CancelCh        CancelStream
	WaitAQ          AnnotatedWgAQ
	Logger          *slog.Logger
}

type PoolResult ¶ added in v0.4.0

type PoolResult struct {
	Error error
}

type PoolResultStream ¶ added in v0.4.0

type PoolResultStream = chan *PoolResult

type PoolResultStreamR ¶ added in v0.4.0

type PoolResultStreamR = <-chan *PoolResult

type PoolResultStreamW ¶ added in v0.4.0

type PoolResultStreamW = chan<- *PoolResult

type WaitGroupAn ¶

type WaitGroupAn interface {
	AnnotatedWgAdder
	AnnotatedWgQuitter
	AnnotatedWgWaiter
	AnnotatedWgCounter
}

WaitGroupAn the extended WaitGroup

func NewAnnotatedWaitGroup(name string, log ...*slog.Logger) WaitGroupAn

type WaitGroupName ¶

type WaitGroupName string

type WorkerPool ¶

type WorkerPool[I, O any] struct {
	RoutineName GoRoutineName
	WaitAQ      AnnotatedWgAQ
	ResultInCh  PoolResultStreamR
	Logger      *slog.Logger
	// contains filtered or unexported fields
}

WorkerPool owns the resultOut channel, because it is the only entity that knows when all workers have completed their work due to the finished channel, which it also owns.

func NewWorkerPool[I, O any](params *NewWorkerPoolParams[I, O]) *WorkerPool[I, O]

func (p *WorkerPool[I, O]) Start(
	parentContext context.Context,
	parentCancel context.CancelFunc,
	outputsChOut chan<- JobOutput[O],
)