func (b *Batch) terminateAll() {
	for i := 0; i < len(b.consumers); i++ {
		b.consumers[i].stop()
	}
	for i := 0; i < len(b.consumers); i++ {
		b.consumers[i].join()
	}
}
terminateAll is going to terminate all consumers if possilbe
package goroutines
import (
	"context"
	"errors"
	"sync"
)
var (
	ErrQueueComplete = errors.New("queue has completed already")
	ErrQueueCTXDone = errors.New("context in queue is done")
)
type Result interface {
	Value() interface{}
	Error() error
}
type BatchFunc func() (interface{}, error)
type Batch struct {
	pool         *Pool
	inputChan    chan *input
	outputChan   chan Result
	retChan      chan Result
	completeChan chan struct{}
	inputOnce    sync.Once
	outputOnce   sync.Once
	resultOnce   sync.Once
	consumers    []*consumer
}
func NewBatch(size int, options ...BatchOption) *Batch {
	o := &batchOption{}
	for _, opt := range options {
		opt(o)
	}
	batchSize := defaultBatchSize
	if o.batchSize > batchSize {
		batchSize = o.batchSize
	}
	b := Batch{
		pool:         NewPool(size),
		inputChan:    make(chan *input, batchSize),
		outputChan:   make(chan Result, batchSize),
		completeChan: make(chan struct{}),
		retChan:      make(chan Result),
	}
	for i := 0; i < size; i++ {
		c := newConsumer()
		b.pool.Schedule(func() {
			defer close(c.closedChan)
			for {
				select {
				case <-c.closeChan:
					return
				default:
				}
				select {
				case <-c.closeChan:
					return
				case in, ok := <-b.inputChan:
					if !ok {
						return
					}
					intf, err := in.fn()
					ret := &result{value: intf, err: err}
					b.outputChan <- ret
				}
			}
		})
		b.consumers = append(b.consumers, c)
	}
	return &b
}
func (b *Batch) Queue(fn BatchFunc) error {
	return b.queue(context.Background(), fn)
}
func (b *Batch) QueueWithContext(ctx context.Context, fn BatchFunc) error {
	return b.queue(ctx, fn)
}
func (b *Batch) QueueComplete() {
	b.inputOnce.Do(func() {
		close(b.completeChan)
		close(b.inputChan)
	})
}
func (b *Batch) Results() <-chan Result {
	b.resultOnce.Do(func() {
		go func() {
			for i := 0; i < len(b.consumers); i++ {
				b.consumers[i].join()
			}
			b.outputOnce.Do(func() {
				close(b.outputChan)
			})
		}()
		go func() {
			for output := range b.outputChan {
				b.retChan <- output
			}
			close(b.retChan)
		}()
	})
	return b.retChan
}
func (b *Batch) WaitAll() {
	for range b.Results() {
	}
}
func (b *Batch) Close() {
	b.QueueComplete()
	b.WaitAll()
	b.terminateAll()
	b.pool.Release()
}
func (b *Batch) GracefulClose() {
	go func() {
		b.Close()
	}()
}
func (b *Batch) queue(ctx context.Context, fn BatchFunc) error {
	select {
	case <-b.completeChan:
		return ErrQueueComplete
	default:
	}
	select {
	case <-b.completeChan:
		return ErrQueueComplete
	case <-ctx.Done():
		return ErrQueueCTXDone
	default:
	}
	select {
	case <-b.completeChan:
		return ErrQueueComplete
	case <-ctx.Done():
		return ErrQueueCTXDone
	case b.inputChan <- &input{fn: fn}:
		return nil
	}
}