func (c Coordinator) Put(ctx context.Context, e Element) ([]Element, error) {
	return c.queue.enqueue(ctx, e)
}
Put new element into inner buffer queue. It return error when inner buffer queue is full, and elements failed putting to queue is the first return value.
Due to different rejectPolicy, multiple elements may be discarded before current element put successfully.
Common usages pattern:

discarded, err := c.Put(e)
if err != nil {
fmt.Errorf("discarded elements %+v for err %v", discarded, err)
}
package prosumer
import (
	"context"
	"log"
	"sync"
	"time"
)
type worker struct {
	*queue
	close     chan struct{}
	waitClose *sync.WaitGroup
	consumer      Consumer
	batchSize     int
	batchInterval time.Duration
	cb            Callback
}
func (w worker) start() {
	received := make([]Element, 0, w.batchSize)
	watchdog := time.NewTimer(w.batchInterval)
	doWork := func() {
		watchdog.Stop()
		if len(received) > 0 {
			err := w.consumer(received)
			w.cb(received, err)
			received = received[:0]
		}
		watchdog.Reset(w.batchInterval)
	}
loop:
	for {
		if e, ok := w.queue.dequeue(); ok {
			received = append(received, e)
			if len(received) >= w.batchSize {
				if enable_race {
					log.Println("doWork for reach batchSize")
				}
				doWork()
			} else {
				select {
				case <-watchdog.C:
					if enable_race {
						log.Println("doWork for reach batchInterval while dequeue")
					}
					doWork()
				default:
				}
			}
		} else {
			select {
			case <-w.close:
				doWork()
				watchdog.Stop()
				w.waitClose.Done()
				break loop
			case <-watchdog.C:
				if enable_race {
					log.Println("doWork for reach batchInterval")
				}
				doWork()
			}
		}
	}
}
type Coordinator struct {
	*queue
	*worker
	numConsumer int
	close     chan struct{}  
	waitClose *sync.WaitGroup
}
func NewCoordinator(config Config) Coordinator {
	var waitClose sync.WaitGroup
	waitClose.Add(config.numConsumer + 1)  
	closeCh := make(chan struct{})
	q := newQueue(config.bufferSize, config.rp, &waitClose)
	c := Coordinator{
		queue: q,
		worker: &worker{
			queue:         q,
			close:         closeCh,
			waitClose:     &waitClose,
			consumer:      config.consumer,
			batchSize:     config.batchSize,
			batchInterval: config.batchInterval,
			cb:            config.cb,
		},
		numConsumer: config.numConsumer,
		close:       closeCh,
		waitClose:   &waitClose,
	}
	return c
}
func (c Coordinator) Start() {
	for i := 0; i < c.numConsumer; i++ {
		go c.worker.start()
	}
}
func (c Coordinator) Close(graceful bool) error {
	c.queue.close(graceful)
	close(c.close)  
	if graceful {
		c.waitClose.Wait()
	}
	return nil
}