conc

Documentation

Index

• type ConcurrentQueue
  • func NewConcurrentQueue(maxConcurrency int, initialElems ...func()) *ConcurrentQueue
  • func (s *ConcurrentQueue) Enqueue(jobs ...func()) (queued, running int)
  • func (s *ConcurrentQueue) WaitIdle(ctx context.Context, errCh <-chan error) error
  • func (s *ConcurrentQueue) WatchState(ctx context.Context, errCh <-chan error, ...) error

Types

type ConcurrentQueue

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

ConcurrentQueue is a pool of goroutines processing a stream of jobs. Job callbacks are called in the order they are added.

func NewConcurrentQueue

func NewConcurrentQueue(maxConcurrency int, initialElems ...func()) *ConcurrentQueue

NewConcurrentQueue constructs a new stream concurrency manager. initialElems contains the initial set of queued entries. if maxConcurrency <= 0, spawns infinite goroutines.

func (*ConcurrentQueue) Enqueue

func (s *ConcurrentQueue) Enqueue(jobs ...func()) (queued, running int)

Enqueue enqueues a job callback to the stream. If possible, the job is started immediately and skips the queue. Returns the current number of queued and running jobs.

func (*ConcurrentQueue) WaitIdle

func (s *ConcurrentQueue) WaitIdle(ctx context.Context, errCh <-chan error) error

WaitIdle waits for no jobs to be running. Returns context.Canceled if ctx is canceled. errCh is an optional error channel.

func (*ConcurrentQueue) WatchState

func (s *ConcurrentQueue) WatchState(
	ctx context.Context,
	errCh <-chan error,
	cb func(queued, running int) (bool, error),
) error

WatchState watches the concurrent queue state. If the callback returns an error or false, returns that error or nil. Returns nil immediately if callback is nil. Returns context.Canceled if ctx is canceled. errCh is an optional error channel.