func (q *queue) Flush() []partitionedItems {
	q.Lock()
	defer q.Unlock()
	flushed := q.queue
	q.queue = make([]partitionedItems, 0, defaultCapacity)
	return flushed
}
Flush flushes the event queue
package async
import (
	"context"
	"reflect"
	"sync"
)
type partitioner struct {
	sync.RWMutex
	ctx       context.Context
	queue     *queue
	partition PartitionFunc  
}
const defaultCapacity = 1 << 14
type partitionedItems map[string][]interface{}
type Partitioner interface {
	Append(items interface{}) Task
	Partition() map[string][]interface{}
}
type PartitionFunc func(data interface{}) (key string, ok bool)
func NewPartitioner(ctx context.Context, partition PartitionFunc) Partitioner {
	return &partitioner{
		ctx:       ctx,
		queue:     newQueue(),
		partition: partition,
	}
}
func (p *partitioner) Append(items interface{}) Task {
	return Invoke(p.ctx, func(context.Context) (interface{}, error) {
		p.queue.Append(p.transform(items))
		return nil, nil
	})
}
func (p *partitioner) transform(items interface{}) partitionedItems {
	t := reflect.TypeOf(items)
	if t.Kind() != reflect.Slice {
		panic("transform requires for slice")
	}
	rv := reflect.ValueOf(items)
	mapped := partitionedItems{}
	for i := 0; i < rv.Len(); i++ {
		e := rv.Index(i).Interface()
		if key, ok := p.partition(e); ok {
			mapped[key] = append(mapped[key], e)
		}
	}
	return mapped
}
func (p *partitioner) Partition() map[string][]interface{} {
	out := partitionedItems{}
	for _, pMap := range p.queue.Flush() {
		for k, v := range pMap {
			out[k] = append(out[k], v...)
		}
	}
	return out
}
type queue struct {
	sync.Mutex
	queue []partitionedItems
}
func newQueue() *queue {
	return &queue{
		queue: make([]partitionedItems, 0, defaultCapacity),
	}
}
func (q *queue) Append(events partitionedItems) {
	q.Lock()
	q.queue = append(q.queue, events)
	q.Unlock()
}