patterns

Documentation

Index

• Constants
• Variables
• func AggregateChannel[T any](ctx context.Context, inputStream <-chan T, f AggregateFunc[T]) <-chan T
• func Detach(ctx context.Context) (context.Context, error)
• func FanIn[T any](sources ...<-chan T) <-chan T
• func FanOut[T any](source <-chan T, n int) []<-chan T
• func FilterChannel[T any](ctx context.Context, inputStream <-chan T, f FilterFunc[T]) <-chan T
• func Pipeline[T any](jobs ...Job[T])
• func Workers[T any](ctx context.Context, f WorkFunc[T], n int, wg *sync.WaitGroup, source <-chan T, ...)
• func Wrap(ctx context.Context, f WrappedFunc) error
• type AggregateFunc
• type Broadcast
  • func NewBroadcast[T any](ctx context.Context, source <-chan T) *Broadcast[T]
  • func (b *Broadcast[T]) Subscribe() <-chan T
  • func (b *Broadcast[T]) SubscribersCount() int
  • func (b *Broadcast[T]) Unsubscribe(channel <-chan T)
• type Broker
  • func NewBroker[T any]() *Broker[T]
  • func (b *Broker[T]) CloseAll() error
  • func (b *Broker[T]) CloseTopic(topic string) error
  • func (b *Broker[T]) CreateTopic(ctx context.Context, topic string) error
  • func (b *Broker[T]) Publish(ctx context.Context, topic string, message T) error
  • func (b *Broker[T]) Subscribe(topic string) (<-chan T, error)
  • func (b *Broker[T]) SubscribersCountAll() (int, error)
  • func (b *Broker[T]) SubscribersCountByTopic(topic string) (int, error)
  • func (b *Broker[T]) Unsubscribe(topic string, ch <-chan T) error
• type Circuit
  • func Breaker[T any](circuit Circuit[T], waitDuration time.Duration, failureThreshold uint) Circuit[T]
  • func Debounce[T any](circuit Circuit[T], d time.Duration) Circuit[T]
• type Effector
  • func Retry[T any](effector Effector[T], retries int, delay time.Duration) Effector[T]
  • func Throttle[T any](e Effector[T], max, refill uint, d time.Duration) Effector[T]
• type FilterFunc
• type Iterator
  • func NewIterator[T any](data []T) *Iterator[T]
  • func (i *Iterator[T]) GetNext() T
  • func (i *Iterator[T]) HasNext() bool
• type Job
• type MutexBucket
  • func NewMutexBucket[T any](elems [SegmentCount]*T) *MutexBucket[T]
  • func (mb *MutexBucket[T]) UnderLock(identifier int64, f func(p *T) error) error
  • func (mb *MutexBucket[T]) UnderReadLock(identifier int64, f func(p *T) error) error
• type Topic
• type WorkFunc
• type WorkersCurator
  • func NewWorkersCurator[T any](ctx context.Context, f WorkFunc[T], source <-chan T, workerCount int) *WorkersCurator[T]
  • func (c *WorkersCurator[T]) GetErrCh() chan error
  • func (c *WorkersCurator[T]) Wait()
• type WrappedFunc

Constants

const (
	SegmentCount = 256
)

Variables

var (
	ErrBrokerClosed       = errors.New("broker closed")
	ErrTopicNotFound      = errors.New("topic not found")
	ErrPublishFailed      = errors.New("publish failed")
	ErrTopicAlreadyExists = errors.New("topic already exists")
)

var (
	ErrCircuitWait = errors.New("err need more time wait before request")
)

var (
	ErrEmptyParentContext = errors.New("empty parent context")
)

var (
	ErrToManyCalls = errors.New("too many calls")
)

Functions

func AggregateChannel

func AggregateChannel[T any](ctx context.Context, inputStream <-chan T, f AggregateFunc[T]) <-chan T

func Detach

func Detach(ctx context.Context) (context.Context, error)

func FanIn

func FanIn[T any](sources ...<-chan T) <-chan T

func FanOut

func FanOut[T any](source <-chan T, n int) []<-chan T

func FilterChannel

func FilterChannel[T any](ctx context.Context, inputStream <-chan T, f FilterFunc[T]) <-chan T

func Pipeline

func Pipeline[T any](jobs ...Job[T])

func Workers

func Workers[T any](
	ctx context.Context,
	f WorkFunc[T],
	n int,
	wg *sync.WaitGroup,
	source <-chan T,
	errCh chan error,
	panicCh chan struct{},
)

func Wrap

func Wrap(ctx context.Context, f WrappedFunc) error

Types

type AggregateFunc

type AggregateFunc[T any] func(data T) T

type Broadcast

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

func NewBroadcast

func NewBroadcast[T any](ctx context.Context, source <-chan T) *Broadcast[T]

func (*Broadcast[T]) Subscribe

func (b *Broadcast[T]) Subscribe() <-chan T

func (*Broadcast[T]) SubscribersCount

func (b *Broadcast[T]) SubscribersCount() int

func (*Broadcast[T]) Unsubscribe

func (b *Broadcast[T]) Unsubscribe(channel <-chan T)

type Broker

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

func NewBroker

func NewBroker[T any]() *Broker[T]

func (*Broker[T]) CloseAll

func (b *Broker[T]) CloseAll() error

func (*Broker[T]) CloseTopic

func (b *Broker[T]) CloseTopic(topic string) error

func (*Broker[T]) CreateTopic

func (b *Broker[T]) CreateTopic(ctx context.Context, topic string) error

func (*Broker[T]) Publish

func (b *Broker[T]) Publish(ctx context.Context, topic string, message T) error

func (*Broker[T]) Subscribe

func (b *Broker[T]) Subscribe(topic string) (<-chan T, error)

func (*Broker[T]) SubscribersCountAll

func (b *Broker[T]) SubscribersCountAll() (int, error)

func (*Broker[T]) SubscribersCountByTopic

func (b *Broker[T]) SubscribersCountByTopic(topic string) (int, error)

func (*Broker[T]) Unsubscribe

func (b *Broker[T]) Unsubscribe(topic string, ch <-chan T) error

type Circuit

type Circuit[T any] func(ctx context.Context) (T, error)

func Breaker

func Breaker[T any](circuit Circuit[T], waitDuration time.Duration, failureThreshold uint) Circuit[T]

func Debounce

func Debounce[T any](circuit Circuit[T], d time.Duration) Circuit[T]

type Effector

type Effector[T any] func(ctx context.Context) (T, error)

func Retry

func Retry[T any](effector Effector[T], retries int, delay time.Duration) Effector[T]

func Throttle

func Throttle[T any](e Effector[T], max, refill uint, d time.Duration) Effector[T]

type FilterFunc

type FilterFunc[T any] func(data T) bool

type Iterator

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

func NewIterator

func NewIterator[T any](data []T) *Iterator[T]

func (*Iterator[T]) GetNext

func (i *Iterator[T]) GetNext() T

func (*Iterator[T]) HasNext

func (i *Iterator[T]) HasNext() bool

type Job

type Job[T any] func(in, out chan T)

type MutexBucket

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

func NewMutexBucket

func NewMutexBucket[T any](elems [SegmentCount]*T) *MutexBucket[T]

func (*MutexBucket[T]) UnderLock

func (mb *MutexBucket[T]) UnderLock(identifier int64, f func(p *T) error) error

func (*MutexBucket[T]) UnderReadLock

func (mb *MutexBucket[T]) UnderReadLock(identifier int64, f func(p *T) error) error

type Topic

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

type WorkFunc

type WorkFunc[T any] func(data T) error

type WorkersCurator

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

func NewWorkersCurator

func NewWorkersCurator[T any](
	ctx context.Context,
	f WorkFunc[T],
	source <-chan T,
	workerCount int,
) *WorkersCurator[T]

func (*WorkersCurator[T]) GetErrCh

func (c *WorkersCurator[T]) GetErrCh() chan error

func (*WorkersCurator[T]) Wait

func (c *WorkersCurator[T]) Wait()

type WrappedFunc

type WrappedFunc func() error