func (b *batch) Size() int {
	b.RLock()
	defer b.RUnlock()
	return len(b.pending)
}
Size returns the length of the pending queue
package async
import (
	"context"
	"errors"
	"fmt"
	"sync"
)
type batchEntry struct {
	id      uint64
	payload interface{}  
	task    Task         
}
type batch struct {
	sync.RWMutex
	ctx       context.Context
	lastID    uint64                             
	pending   []batchEntry                       
	batchTask Task                               
	batch     chan []batchEntry                  
	process   func([]interface{}) []interface{}  
}
type Batch interface {
	Append(payload interface{}) Task
	Size() int
	Reduce()
}
func NewBatch(ctx context.Context, process func([]interface{}) []interface{}) Batch {
	return &batch{
		ctx:     ctx,
		pending: []batchEntry{},
		batch:   make(chan []batchEntry),
		process: process,
	}
}
func (b *batch) Append(payload interface{}) Task {
	b.Lock()
	defer b.Unlock()
	b.lastID = b.lastID + 1
	id := b.lastID
	if b.batchTask == nil {
		b.batchTask = b.createBatchTask()
	}
	t := b.batchTask.ContinueWith(b.ctx, func(batchResult interface{}, _ error) (interface{}, error) {
		if res, ok := batchResult.(map[uint64]interface{}); ok {
			return res[id], nil
		}
		actualType := fmt.Sprintf("%T", batchResult)
		return nil, errors.New("Invalid batch type, got: " + actualType)
	})
	b.pending = append(b.pending, batchEntry{
		id:      id,
		payload: payload,
		task:    t,
	})
	return t
}
func (b *batch) Reduce() {
	b.Lock()
	defer b.Unlock()
	if len(b.pending) == 0 {
		return
	}
	batch := append([]batchEntry{}, b.pending...)
	b.batch <- batch
	b.batchTask = b.createBatchTask()
}