func (b *Bundler) Flush() {
	b.mu.Lock()
	b.closeBundle()
	select {
	case b.handlec <- 1:
	default:
	}
	calledc := b.calledc  
	b.mu.Unlock()
	<-calledc
}
Flush waits until all items in the Bundler have been handled (that is,
until the last invocation of handler has returned).
package bundler
import (
	"errors"
	"reflect"
	"sync"
	"time"
)
const (
	DefaultDelayThreshold       = time.Second
	DefaultBundleCountThreshold = 10
	DefaultBundleByteThreshold  = 1e6  
	DefaultBufferedByteLimit    = 1e9  
)
var (
	ErrOverflow = errors.New("bundler reached buffered byte limit")
	ErrOversizedItem = errors.New("item size exceeds bundle byte limit")
)
type Bundler struct {
	DelayThreshold time.Duration
	BundleCountThreshold int
	BundleByteThreshold int
	BundleByteLimit int
	BufferedByteLimit int
	handler       func(interface{})  
	itemSliceZero reflect.Value      
	donec         chan struct{}      
	handlec       chan int           
	timer         *time.Timer        
	mu            sync.Mutex
	bufferedSize  int            
	closedBundles []bundle       
	curBundle     bundle         
	calledc       chan struct{}  
}
type bundle struct {
	items reflect.Value  
	size  int            
}
func NewBundler(itemExample interface{}, handler func(interface{})) *Bundler {
	b := &Bundler{
		DelayThreshold:       DefaultDelayThreshold,
		BundleCountThreshold: DefaultBundleCountThreshold,
		BundleByteThreshold:  DefaultBundleByteThreshold,
		BufferedByteLimit:    DefaultBufferedByteLimit,
		handler:       handler,
		itemSliceZero: reflect.Zero(reflect.SliceOf(reflect.TypeOf(itemExample))),
		donec:         make(chan struct{}),
		handlec:       make(chan int, 1),
		calledc:       make(chan struct{}),
		timer:         time.NewTimer(1000 * time.Hour),  
	}
	b.curBundle.items = b.itemSliceZero
	go b.background()
	return b
}
func (b *Bundler) Add(item interface{}, size int) error {
	if b.BundleByteLimit > 0 && size > b.BundleByteLimit {
		return ErrOversizedItem
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.bufferedSize+size > b.BufferedByteLimit {
		return ErrOverflow
	}
	if b.BundleByteLimit > 0 && b.curBundle.size+size > b.BundleByteLimit {
		b.closeAndHandleBundle()
	}
	b.curBundle.items = reflect.Append(b.curBundle.items, reflect.ValueOf(item))
	b.curBundle.size += size
	b.bufferedSize += size
	if b.curBundle.items.Len() == 1 {
		b.timer.Reset(b.DelayThreshold)
	}
	if b.curBundle.items.Len() == b.BundleCountThreshold {
		b.closeAndHandleBundle()
	}
	if b.curBundle.size >= b.BundleByteThreshold {
		b.closeAndHandleBundle()
	}
	return nil
}