patterns

Dec 19, 2024

Changes in this version

+ const ChanceEighth

+ const ChanceHalf

+ const ChanceQuarter

+ type BoolGenerator struct

+ func NewBoolGenerator(c Chance) *BoolGenerator

+ func (bg *BoolGenerator) Bool() bool

+ type Chance string

Jul 31, 2024

Changes in this version

type Queue

+ func (q *Queue[T, M]) PopByChannel(reqTime time.Duration) (<-chan NodeInfo[T, M], error)

Jul 30, 2024

Changes in this version

+ const MaxPriority

+ const MinPriority

+ var ErrEmptyChannel = errors.New("channel is empty")

+ var ErrEmptyQueue = errors.New("queue is empty")

+ var ErrEmptyReqTime = errors.New("request time is empty")

+ var ErrKeyDoesNotExist = errors.New("key does not exist")

+ type NodeInfo struct

+ Key M

+ Priority int64

+ Value T

+ type NodeInfoChannel chan NodeInfo[T, M]

+ type Queue struct

+ func NewQueue[T any, M comparable](opts ...QueueOption[T, M]) *Queue[T, M]

+ func (q *Queue[T, M]) Close()

+ func (q *Queue[T, M]) Pop() (NodeInfo[T, M], error)

+ func (q *Queue[T, M]) PopByChannelWithReqTime(reqTime time.Duration) (<-chan NodeInfo[T, M], error)

+ func (q *Queue[T, M]) PopByKey(key M) (NodeInfo[T, M], error)

+ func (q *Queue[T, M]) Push(priority int64, key M, value T)

+ func (q *Queue[T, M]) Size() uint64

+ type QueueOption func(q *Queue[T, M])

+ func WithChannel[T any, M comparable]() QueueOption[T, M]

Feb 23, 2024

Changes in this version

+ type MutexBucketFunc func(p *T) error

Feb 22, 2024

Changes in this version

+ var ErrEmptySingleFlight = errors.New("empty singleflight")

+ var ErrFailedTypeAssert = errors.New("failed type assert")

+ func DoChanCtx[T any](ctx context.Context, sf *singleflight.Group, key string, ...) (<-chan Result[T], error)

+ func DoChan[T any](sf *singleflight.Group, key string, fn func() (any, error)) (<-chan Result[T], error)

+ func Do[T any](sf *singleflight.Group, key string, fn func() (any, error)) (v T, err error, shared bool)

+ func Forget(sf *singleflight.Group, key string) error

+ func TickerV2(ctx context.Context, interval time.Duration, f TickerFunc)

+ type Result struct

+ Err error

+ Shared bool

+ Val T

Aug 19, 2023

Changes in this version

+ var ErrInvalidCount = errors.New("invalid count")

+ type Semaphore struct

+ C chan struct{}

+ func NewSemaphore(n int) (*Semaphore, error)

+ func (s *Semaphore) Acquire()

+ func (s *Semaphore) AcquireCtx(ctx context.Context) error

+ func (s *Semaphore) Release()

Aug 7, 2023

Changes in this version

+ func Ticker(ctx context.Context, interval time.Duration, f TickerFunc)

+ type TickerFunc func()

Jun 16, 2023

Changes in this version

+ type CacheMap struct

+ func NewCacheMap(f ReloadFunc[K, T], clearTick time.Duration) *CacheMap[K, T]

+ func (cm *CacheMap[K, T]) ClearCache()

+ func (cm *CacheMap[K, T]) Get(key K) (T, bool)

+ func (cm *CacheMap[K, T]) RemoveCache()

+ func (cm *CacheMap[K, T]) Set(key K, value T)

+ type ReloadFunc func(oldCache map[K]T) map[K]T

May 30, 2023

Changes in this version

+ const SegmentCount

+ var ErrBrokerClosed = errors.New("broker closed")

+ var ErrPublishFailed = errors.New("publish failed")

+ var ErrTopicAlreadyExists = errors.New("topic already exists")

+ var ErrTopicNotFound = errors.New("topic not found")

type Broadcast

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

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

+ type Broker struct

+ func NewBroker() *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 MutexBucket struct

+ func NewMutexBucket(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 struct

May 9, 2023

Changes in this version

+ func FanIn(sources ...<-chan T) <-chan T

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

May 7, 2023

Changes in this version

+ 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")

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

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

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

+ func Funnel(sources ...<-chan T) <-chan T

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

+ func Split(source <-chan T, n int) []<-chan T

+ func Workers(ctx context.Context, f WorkFunc[T], n int, wg *sync.WaitGroup, source <-chan T, ...)

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

+ type AggregateFunc func(data T) T

+ type Broadcast struct

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

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

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

+ type Circuit func(ctx context.Context) (T, error)

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

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

+ type Effector func(ctx context.Context) (T, error)

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

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

+ type FilterFunc func(data T) bool

+ type Iterator struct

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

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

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

+ type Job func(in, out chan T)

+ type WorkFunc func(data T) error

+ type WorkersCurator struct

+ func NewWorkersCurator(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 func() error