func (q *retryQueue) Front() []byte {
	now := time.Now()
	for {
		el := q.batches.Front()
		if el == nil {
			return nil
		}
		batch := el.Value.(retryBatch)
		if now.After(batch.ttl) {
			q.batches.Remove(el)
		} else {
			return batch.buf
		}
	}
}
Front returns the buffer at the start of the queue. Nil is returned
if the queue is empty.
package writer
import (
	"container/list"
	"sync"
	"time"
	"github.com/jumptrading/influx-spout/v2/config"
)
type dbWriter interface {
	Write([]byte) error
}
func newRetryWorker(inputCh <-chan []byte, dbWriter dbWriter, c *config.Config) *retryWorker {
	w := &retryWorker{
		inputCh:       inputCh,
		dbWriter:      dbWriter,
		retryInterval: c.WriterRetryInterval.Duration,
		queue: &retryQueue{
			maxBatches: c.WriterRetryBatches,
			maxTTL:     c.WriterRetryTimeout.Duration,
		},
		writes:      make(chan struct{}, c.WriterRetryBatches*4),
		writeErrors: make(chan error, c.WriterRetryBatches*4),
		stop:        make(chan struct{}),
	}
	w.wg.Add(1)
	go w.loop()
	return w
}
type retryWorker struct {
	inputCh       <-chan []byte
	dbWriter      dbWriter
	retryInterval time.Duration
	writes        chan struct{}
	writeErrors   chan error
	queue         *retryQueue
	stop          chan struct{}
	wg            sync.WaitGroup
}
func (w *retryWorker) Stop() {
	close(w.stop)
	w.wg.Wait()
}
func (w *retryWorker) WriteErrors() <-chan error {
	return w.writeErrors
}
func (w *retryWorker) Writes() <-chan struct{} {
	return w.writes
}
func (w *retryWorker) loop() {
	defer w.wg.Done()
	var nextSend <-chan time.Time
	for {
		if nextSend == nil && w.queue.Len() > 0 {
			nextSend = time.After(w.retryInterval)
		}
		select {
		case <-w.stop:
			return
		case buf := <-w.inputCh:
			w.queue.Add(buf)
		case <-nextSend:
			nextSend = nil
			buf := w.queue.Front()
			if buf != nil {
				w.reportWrite()
				err := w.dbWriter.Write(buf)
				if err != nil {
					if isPermanentError(err) {
						w.queue.DropFront()
					} else {
						w.queue.CycleFront()
					}
					w.reportError(err)
				} else {
					w.queue.DropFront()
				}
			}
		}
	}
}
func (w *retryWorker) reportWrite() {
	select {
	case w.writes <- struct{}{}:
	default:
	}
}
func (w *retryWorker) reportError(err error) {
	select {
	case w.writeErrors <- err:
	default:
	}
}
type retryQueue struct {
	maxTTL     time.Duration
	maxBatches int
	batches    list.List
}
func (q *retryQueue) Len() int {
	return q.batches.Len()
}
func (q *retryQueue) Add(batch []byte) {
	if q.batches.Len() >= q.maxBatches {
		q.removeOldest()
	}
	q.batches.PushBack(retryBatch{
		buf: batch,
		ttl: time.Now().Add(q.maxTTL),
	})
}