worker

Documentation

Index

• type FanOut
  • func NewFanOut[I any](workerCount, bufferSize uint) *FanOut[I]
  • func (f *FanOut[I]) Close()
  • func (f *FanOut[I]) Invoke(input I)
  • func (f *FanOut[I]) Process(p ProcessorFunc[I])
• type HashFunc
  • func StringHasher[I any](keyFunc KeyFunc[I, string]) HashFunc[I]
• type HashedFanOut
  • func NewHashedFanOut[I any](workerCount, bufferSize uint, hasher HashFunc[I]) *HashedFanOut[I]
  • func (f *HashedFanOut[I]) Close()
  • func (f *HashedFanOut[I]) Invoke(input I)
  • func (f *HashedFanOut[I]) Process(p ProcessorFunc[I])
• type KeyFunc
• type ProcessorFunc

Types

type FanOut

type FanOut[I any] struct {
	// contains filtered or unexported fields
}

FanOut will fan out to `workerCount` total coroutines for processing.

func NewFanOut

func NewFanOut[I any](workerCount, bufferSize uint) *FanOut[I]

func (*FanOut[I]) Close

func (f *FanOut[I]) Close()

Close closes the input channels. Invoke can not be called again after this call.

func (*FanOut[I]) Invoke

func (f *FanOut[I]) Invoke(input I)

Invoke adds the data to the worker for processing.

func (*FanOut[I]) Process

func (f *FanOut[I]) Process(p ProcessorFunc[I])

Process handles all inputs until the input channel is closed.

type HashFunc

type HashFunc[I any] func(I) uint

HashFunc converts an input to a uint value. It must be deterministic. See examples below.

func StringHasher

func StringHasher[I any](keyFunc KeyFunc[I, string]) HashFunc[I]

StringHasher given a KeyFunc that returns a string for a given input, returns a consistent hash for the string.

type HashedFanOut

type HashedFanOut[I any] struct {
	// contains filtered or unexported fields
}

HashedFanOut is a special purpose fan out that hashes the input so that the same keys are always processed by the same worker. This is used to ensure guard against race conditions in non-reentrant processors.

func NewHashedFanOut

func NewHashedFanOut[I any](workerCount, bufferSize uint, hasher HashFunc[I]) *HashedFanOut[I]

func (*HashedFanOut[I]) Close

func (f *HashedFanOut[I]) Close()

func (*HashedFanOut[I]) Invoke

func (f *HashedFanOut[I]) Invoke(input I)

func (*HashedFanOut[I]) Process

func (f *HashedFanOut[I]) Process(p ProcessorFunc[I])

type KeyFunc

type KeyFunc[I, K any] func(I) K

type ProcessorFunc

type ProcessorFunc[I any] func(I)