func NewBatchRunnerWithConfig(
	ctx context.Context,
	config BatchConfig,
) *BatchRunner {
	if config.Concurrency == 0 {
		config.Concurrency = defaultConcurrency
	}
	runner := BatchRunner{
		config: config,
		argCh:  make([]chan argument, config.Concurrency),
		resCh:  make(chan Result, 64),
	}
	runner.wg.Add(int(config.Concurrency))
	for i := 0; i < int(config.Concurrency); i++ {
		runner.argCh[i] = make(chan argument, 64)
		go runner.run(ctx, runner.argCh[i])
	}
	return &runner
}
NewBatchRunnerWithConfig creates a new BatchRunner using the configuration
parameters provided in BatchConfig. If the ctx is Done the
BatchRunner execution stops
package concurrent
import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	stderr "github.com/pkg/errors"
)
type ErrNoOccurrence struct{}
func (e ErrNoOccurrence) Error() string {
	return "the expected event never occurred"
}
type ErrCannotRecover struct {
	Cause error
}
func (e ErrCannotRecover) Error() string {
	return e.Cause.Error()
}
type ErrMaxAttemptsReached struct {
	Causes []error
}
func (e ErrMaxAttemptsReached) Error() string {
	return fmt.Sprintf("maximum number of attempts %d reached", len(e.Causes))
}
const (
	defaultConcurrency     uint8         = 2
	defaultBaseTimeout     time.Duration = 100 * time.Millisecond
	defaultBaseExp         uint8         = 2
	defaultMaxRetryTimeout time.Duration = 10 * time.Second
	defaultAttempts        uint8         = 10
)
var RandomConfig = RetryConfig{
	BaseTimeout:     defaultBaseTimeout,
	BaseExp:         defaultBaseExp,
	MaxRetryTimeout: defaultMaxRetryTimeout,
	Attempts:        defaultAttempts,
	Random:          true,
}
type Supplier interface {
	Supply() (interface{}, error)
}
type SupplierFunc func() (interface{}, error)
func (s SupplierFunc) Supply() (interface{}, error) {
	return s()
}
type RetryConfig struct {
	Random bool
	UnlimitedAttempts bool
	Attempts uint8
	BaseExp uint8
	BaseTimeout time.Duration
	MaxRetryTimeout time.Duration
}
func RetryWithConfig(
	ctx context.Context,
	supplier Supplier,
	config RetryConfig,
) (interface{}, error) {
	var errs []error
	timeout := config.BaseTimeout.Nanoseconds()
	exp := int64(config.BaseExp)
	maxTimeout := config.MaxRetryTimeout.Nanoseconds()
	attempts := 0
	maxAttempts := int(config.Attempts)
	timer := time.NewTimer(0 * time.Second)
	if config.UnlimitedAttempts {
		maxAttempts = -1
	}
	for {
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil, stderr.WithStack(context.Canceled)
		case <-timer.C:
			v, err := supplier.Supply()
			if err == nil {
				return v, nil
			}
			if err, ok := err.(ErrCannotRecover); ok {
				return nil, stderr.WithStack(err.Cause)
			}
			errs = append(errs, err)
		}
		attempts++
		if attempts >= maxAttempts && maxAttempts >= 0 {
			return nil, ErrMaxAttemptsReached{Causes: errs}
		}
		timeout = (timeout * exp)
		multiplier := rand.Float64() + 0.5
		if timeout > maxTimeout {
			timeout = maxTimeout
			multiplier = rand.Float64() + 1
		}
		if config.Random {
			timeout = int64(multiplier*float64(timeout)) + 1
		}
		timer.Reset(time.Duration(timeout))
	}
}
func Retry(ctx context.Context, supplier Supplier) (interface{}, error) {
	return RetryWithConfig(ctx, supplier, RetryConfig{
		BaseTimeout:     defaultBaseTimeout,
		BaseExp:         defaultBaseExp,
		MaxRetryTimeout: defaultMaxRetryTimeout,
		Attempts:        defaultAttempts,
	})
}
func RetryRandom(ctx context.Context, supplier Supplier) (interface{}, error) {
	return RetryWithConfig(ctx, supplier, RandomConfig)
}
type PoolConfig struct {
	Concurrency uint8
}
type PoolRunner struct {
	config  PoolConfig
	wg      sync.WaitGroup
	argCh   []chan argument
	counter uint64
}
func NewPoolRunner(ctx context.Context) *PoolRunner {
	return NewPoolRunnerWithConfig(ctx, PoolConfig{
		Concurrency: defaultConcurrency,
	})
}
func NewPoolRunnerWithConfig(ctx context.Context, config PoolConfig) *PoolRunner {
	if config.Concurrency == 0 {
		config.Concurrency = defaultConcurrency
	}
	runner := PoolRunner{
		config: config,
		argCh:  make([]chan argument, config.Concurrency),
	}
	runner.wg.Add(int(config.Concurrency))
	for i := 0; i < int(config.Concurrency); i++ {
		runner.argCh[i] = make(chan argument, 128)
		go runner.run(ctx, runner.argCh[i])
	}
	return &runner
}
func (r *PoolRunner) run(ctx context.Context, inCh <-chan argument) {
	go func() {
		defer r.wg.Done()
		for {
			select {
			case <-ctx.Done():
				return
			case arg, ok := <-inCh:
				if !ok {
					return
				}
				s := arg.Supplier
				arg.TimeExecuted = time.Now().UnixNano()
				v, err := s.Supply()
				arg.TimeDone = time.Now().UnixNano()
				if arg.Out != nil {
					arg.Out <- Result{
						Result:        v,
						Err:           err,
						Index:         arg.Index,
						TimeSubmitted: arg.TimeSubmitted,
						TimeExecuted:  arg.TimeExecuted,
						TimeDone:      arg.TimeDone,
					}
				}
			}
		}
	}()
}
func (r *PoolRunner) Run(out chan<- Result, supplier Supplier) {
	if out == nil {
		panic("channel cannot be nil")
	}
	current := atomic.AddUint64(&r.counter, 1)
	index := r.counter % uint64(r.config.Concurrency)
	r.argCh[index] <- argument{
		Out:           out,
		Supplier:      supplier,
		Index:         current,
		TimeSubmitted: time.Now().UnixNano(),
	}
}
func (r *PoolRunner) RunAndDiscard(supplier Supplier) {
	current := atomic.AddUint64(&r.counter, 1)
	index := r.counter % uint64(r.config.Concurrency)
	r.argCh[index] <- argument{
		Out:           nil,
		Supplier:      supplier,
		Index:         current,
		TimeSubmitted: time.Now().UnixNano(),
	}
}
func (r *PoolRunner) Stop() {
	for _, ch := range r.argCh {
		close(ch)
	}
	r.wg.Wait()
}
type BatchRunner struct {
	config BatchConfig
	wg     sync.WaitGroup
	argCh  []chan argument
	resCh  chan Result
}
func NewBatchRunner(ctx context.Context) *BatchRunner {
	return NewBatchRunnerWithConfig(ctx, BatchConfig{Concurrency: defaultConcurrency})
}