pubsub

Aug 29, 2023

Changes in this version

type Distributor

+ func (d Distributor[T]) WithInputFilter(filter func(T) bool) Distributor[T]

+ func (d Distributor[T]) WithOutputFilter(filter func(T) bool) Distributor[T]

Aug 14, 2023

Changes in this version

type Broker

+ func (b *Broker[T]) Populate(iter *fun.Iterator[T]) fun.Worker

type Distributor

+ func DistributorChanOp(chop fun.ChanOp[T]) Distributor[T]

Jul 7, 2023

Changes in this version

type Deque

+ func (dq *Deque[T]) DistributorNonBlocking() Distributor[T]

+ func (dq *Deque[T]) Producer() fun.Producer[T]

+ func (dq *Deque[T]) ProducerBlocking() fun.Producer[T]

+ func (dq *Deque[T]) ProducerReverse() fun.Producer[T]

+ func (dq *Deque[T]) ProducerReverseBlocking() fun.Producer[T]

type Distributor

+ func MakeDistributor(processor fun.Processor[T], producer fun.Producer[T], length func() int) Distributor[T]

+ func (d Distributor[T]) Processor() fun.Processor[T]

+ func (d Distributor[T]) Producer() fun.Producer[T]

type Queue

+ func (q *Queue[T]) Producer() fun.Producer[T]

May 18, 2023

Changes in this version

type Deque

+ func NewUnlimitedDeque() *Deque[T]

+ func (dq *Deque[T]) Distributor() Distributor[T]

+ func (dq *Deque[T]) DistributorLIFO() Distributor[T]

type Distributor

+ func (d Distributor[T]) Iterator() fun.Iterator[T]

type Queue

+ func (q *Queue[T]) Distributor() Distributor[T]

Mar 18, 2023

Changes in this version

type Broker

+ func (b *Broker[T]) Stats(ctx context.Context) BrokerStats

+ type BrokerStats struct

+ BufferDepth int

+ Subscriptions int

type Distributor

+ Len func() int

type Queue

+ func (q *Queue[T]) Len() int

Feb 28, 2023

Changes in this version

+ func DistributorIterator(dist Distributor[T]) fun.Iterator[T]

type Broker

+ func MakeDistributorBroker(ctx context.Context, dist Distributor[T], opts BrokerOptions) *Broker[T]

+ type Distributor interface

+ Receive func(context.Context) (T, error)

+ Send func(context.Context, T) error

+ func DistributorBuffer(d *Deque[T]) Distributor[T]

+ func DistributorChannel(ch chan T) Distributor[T]

+ func DistributorDeque(d *Deque[T]) Distributor[T]

+ func DistributorLIFO(d *Deque[T]) Distributor[T]

+ func DistributorQueue(q *Queue[T]) Distributor[T]

Feb 23, 2023

Changes in this version

type Deque

+ func (dq *Deque[T]) WaitPushBack(ctx context.Context, it T) error

+ func (dq *Deque[T]) WaitPushFront(ctx context.Context, it T) error

type Queue

+ func (q *Queue[T]) BlockingAdd(ctx context.Context, item T) error

Jan 23, 2023

Changes in this version

+ func Populate(ctx context.Context, iter fun.Iterator[T], broker *Broker[T]) error

Jan 21, 2023

Changes in this version

+ var ErrConfigurationMalformed = errors.New("configuration error")

+ var ErrQueueClosed = errors.New("queue is closed")

+ var ErrQueueFull = errors.New("queue is full")

+ var ErrQueueNoCredit = errors.New("insufficient burst credit")

+ type Broker struct

+ func NewBroker(ctx context.Context, opts BrokerOptions) *Broker[T]

+ func NewDequeBroker(ctx context.Context, opts BrokerOptions, deque *Deque[T]) *Broker[T]

+ func NewLIFOBroker(ctx context.Context, opts BrokerOptions, capacity int) (*Broker[T], error)

+ func NewQueueBroker(ctx context.Context, opts BrokerOptions, queue *Queue[T]) *Broker[T]

+ func (b *Broker[T]) Publish(ctx context.Context, msg T)

+ func (b *Broker[T]) Stop()

+ func (b *Broker[T]) Subscribe(ctx context.Context) chan T

+ func (b *Broker[T]) Unsubscribe(ctx context.Context, msgCh chan T)

+ func (b *Broker[T]) Wait(ctx context.Context)

+ type BrokerOptions struct

+ BufferSize int

+ NonBlockingSubscriptions bool

+ ParallelDispatch bool

+ WorkerPoolSize int

+ type Deque struct

+ func NewDeque(opts DequeOptions) (*Deque[T], error)

+ func (dq *Deque[T]) Close() error

+ func (dq *Deque[T]) ForcePushBack(it T) error

+ func (dq *Deque[T]) ForcePushFront(it T) error

+ func (dq *Deque[T]) Iterator() fun.Iterator[T]

+ func (dq *Deque[T]) IteratorBlocking() fun.Iterator[T]

+ func (dq *Deque[T]) IteratorBlockingReverse() fun.Iterator[T]

+ func (dq *Deque[T]) IteratorReverse() fun.Iterator[T]

+ func (dq *Deque[T]) Len() int

+ func (dq *Deque[T]) PopBack() (T, bool)

+ func (dq *Deque[T]) PopFront() (T, bool)

+ func (dq *Deque[T]) PushBack(it T) error

+ func (dq *Deque[T]) PushFront(it T) error

+ func (dq *Deque[T]) WaitBack(ctx context.Context) (T, error)

+ func (dq *Deque[T]) WaitFront(ctx context.Context) (T, error)

+ type DequeOptions struct

+ Capacity int

+ QueueOptions *QueueOptions

+ Unlimited bool

+ func (opts *DequeOptions) Validate() error

+ type Queue struct

+ func NewQueue(opts QueueOptions) (*Queue[T], error)

+ func NewUnlimitedQueue() *Queue[T]

+ func (q *Queue[T]) Add(item T) error

+ func (q *Queue[T]) Close() error

+ func (q *Queue[T]) Iterator() fun.Iterator[T]

+ func (q *Queue[T]) Remove() (T, bool)

+ func (q *Queue[T]) Wait(ctx context.Context) (T, error)

+ type QueueOptions struct

+ BurstCredit float64

+ HardLimit int

+ SoftQuota int

+ func (opts *QueueOptions) Validate() error