func (w *Worker) ExportMetrics() (*Metrics, error) {
	var queueMetrics []*QueueMetrics
	for _, h := range w.handlerMap {
		queue := w.opt.Queue
		ns := w.opt.Namespace
		exporter, ok := queue.(MetricsExporter)
		if !ok {
			continue
		}
		m, err := exporter.GetQueueMetrics(&QueueMetricsOptions{
			Namespace: ns,
			QueueID:   h.QueueID,
			At:        time.Now(),
		})
		if err != nil {
			return nil, err
		}
		queueMetrics = append(queueMetrics, m)
	}
	return &Metrics{
		Queue: queueMetrics,
	}, nil
}
ExportMetrics dumps queue stats if the queue implements MetricsExporter.
package work
import (
	"context"
	"errors"
	"fmt"
	"runtime/debug"
	"sync"
	"time"
	"github.com/cenkalti/backoff/v4"
)
type DequeueFunc func(*DequeueOptions) (*Job, error)
type DequeueMiddleware func(DequeueFunc) DequeueFunc
type EnqueueFunc func(*Job, *EnqueueOptions) error
type EnqueueMiddleware func(EnqueueFunc) EnqueueFunc
type HandleFunc func(*Job, *DequeueOptions) error
type ContextHandleFunc func(context.Context, *Job, *DequeueOptions) error
type HandleMiddleware func(HandleFunc) HandleFunc
type BackoffFunc func(*Job, *DequeueOptions) time.Duration
type handler struct {
	QueueID    string
	HandleFunc ContextHandleFunc
	JobOptions JobOptions
}
type WorkerOptions struct {
	Namespace string
	Queue     Queue
	ErrorFunc func(error)
}
type Worker struct {
	opt WorkerOptions
	cancel     func()
	wg         sync.WaitGroup
	handlerMap map[string]handler
}
func NewWorker(opt *WorkerOptions) *Worker {
	return &Worker{
		opt:        *opt,
		handlerMap: make(map[string]handler),
	}
}
type JobOptions struct {
	MaxExecutionTime time.Duration
	IdleWait         time.Duration
	NumGoroutines    int64
	Backoff          BackoffFunc
	DequeueMiddleware []DequeueMiddleware
	HandleMiddleware  []HandleMiddleware
}
func (opt *JobOptions) AddDequeueMiddleware(mw DequeueMiddleware) *JobOptions {
	opt.DequeueMiddleware = append(opt.DequeueMiddleware, mw)
	return opt
}
func (opt *JobOptions) AddHandleMiddleware(mw HandleMiddleware) *JobOptions {
	opt.HandleMiddleware = append(opt.HandleMiddleware, mw)
	return opt
}
var (
	ErrMaxExecutionTime = errors.New("work: max execution time should be > 0")
	ErrNumGoroutines    = errors.New("work: number of goroutines should be > 0")
	ErrIdleWait         = errors.New("work: idle wait should be > 0")
)
func (opt *JobOptions) Validate() error {
	if opt.MaxExecutionTime <= 0 {
		return ErrMaxExecutionTime
	}
	if opt.IdleWait <= 0 {
		return ErrIdleWait
	}
	if opt.NumGoroutines <= 0 {
		return ErrNumGoroutines
	}
	return nil
}
func (w *Worker) Register(queueID string, h HandleFunc, opt *JobOptions) error {
	return w.RegisterWithContext(queueID, func(ctx context.Context, job *Job, o *DequeueOptions) error {
		return h(job, o)
	}, opt)
}
func (w *Worker) RegisterWithContext(queueID string, h ContextHandleFunc, opt *JobOptions) error {
	err := opt.Validate()
	if err != nil {
		return err
	}
	w.handlerMap[queueID] = handler{
		QueueID: queueID,
		HandleFunc: func(ctx context.Context, job *Job, o *DequeueOptions) error {
			ctx, cancel := context.WithTimeout(ctx, opt.MaxExecutionTime)
			defer cancel()
			return h(ctx, job, o)
		},
		JobOptions: *opt,
	}
	return nil
}
type OnceJobOptions struct {
	MaxExecutionTime time.Duration
	Backoff          BackoffFunc
	DequeueMiddleware []DequeueMiddleware
	HandleMiddleware  []HandleMiddleware
}
func (opt *OnceJobOptions) AddDequeueMiddleware(mw DequeueMiddleware) *OnceJobOptions {
	opt.DequeueMiddleware = append(opt.DequeueMiddleware, mw)
	return opt
}
func (opt *OnceJobOptions) AddHandleMiddleware(mw HandleMiddleware) *OnceJobOptions {
	opt.HandleMiddleware = append(opt.HandleMiddleware, mw)
	return opt
}
func (opt *OnceJobOptions) Validate() error {
	if opt.MaxExecutionTime <= 0 {
		return ErrMaxExecutionTime
	}
	return nil
}
func (w *Worker) RunOnce(ctx context.Context, queueID string, h ContextHandleFunc, opt *OnceJobOptions) error {
	err := opt.Validate()
	if err != nil {
		return err
	}
	queue := w.opt.Queue
	ns := w.opt.Namespace
	dequeue := queue.Dequeue
	for _, mw := range opt.DequeueMiddleware {
		dequeue = mw(dequeue)
	}
	handle := func(job *Job, o *DequeueOptions) error {
		ctx, cancel := context.WithTimeout(ctx, opt.MaxExecutionTime)
		defer cancel()
		return h(ctx, job, o)
	}
	for _, mw := range opt.HandleMiddleware {
		handle = mw(handle)
	}
	handle = catchPanic(handle)
	b := opt.Backoff
	if b == nil {
		b = defaultBackoff()
	}
	handle = retry(queue, b)(handle)
	dopt := &DequeueOptions{
		Namespace:    ns,
		QueueID:      queueID,
		At:           time.Now(),
		InvisibleSec: int64(opt.MaxExecutionTime / time.Second),
	}
	job, err := dequeue(dopt)
	if err != nil {
		return err
	}
	err = handle(job, dopt)
	if err != nil {
		return err
	}
	err = queue.Ack(job, &AckOptions{
		Namespace: dopt.Namespace,
		QueueID:   dopt.QueueID,
	})
	if err != nil {
		return err
	}
	return nil
}
func (w *Worker) Start() {
	ctx, cancel := context.WithCancel(context.Background())
	w.cancel = cancel
	for _, h := range w.handlerMap {
		for i := int64(0); i < h.JobOptions.NumGoroutines; i++ {
			w.wg.Add(1)
			go w.start(ctx, h)
		}
	}
}
func (w *Worker) start(ctx context.Context, h handler) {
	defer w.wg.Done()
	queue := w.opt.Queue
	ns := w.opt.Namespace
	errFunc := func(err error) { fmt.Println(err) }
	if w.opt.ErrorFunc != nil {
		errFunc = w.opt.ErrorFunc
	}
	dequeue := getDequeueFunc(queue)
	for _, mw := range h.JobOptions.DequeueMiddleware {
		dequeue = mw(dequeue)
	}
	dequeue = idleWait(ctx, h.JobOptions.IdleWait)(dequeue)
	handle := func(job *Job, o *DequeueOptions) error {
		return h.HandleFunc(ctx, job, o)
	}
	for _, mw := range h.JobOptions.HandleMiddleware {
		handle = mw(handle)
	}
	handle = catchPanic(handle)
	b := h.JobOptions.Backoff
	if b == nil {
		b = defaultBackoff()
	}
	handle = retry(queue, b)(handle)
	var ackJobs []*Job
	flush := func() error {
		opt := &AckOptions{
			Namespace: ns,
			QueueID:   h.QueueID,
		}
		bulkDeq, ok := queue.(BulkDequeuer)
		if ok {
			err := bulkDeq.BulkAck(ackJobs, opt)
			if err != nil {
				return err
			}
			ackJobs = nil
			return nil
		}
		for _, job := range ackJobs {
			err := queue.Ack(job, opt)
			if err != nil {
				return err
			}
		}
		ackJobs = nil
		return nil
	}
	defer func() {
		err := flush()
		if err != nil {
			errFunc(err)
		}
	}()
	const flushIntv = time.Second
	flushTicker := time.NewTicker(flushIntv)
	defer flushTicker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-flushTicker.C:
			err := flush()
			if err != nil {
				errFunc(err)
			}
		default:
			func() error {
				opt := &DequeueOptions{
					Namespace:    ns,
					QueueID:      h.QueueID,
					At:           time.Now(),
					InvisibleSec: int64((h.JobOptions.MaxExecutionTime + flushIntv) / time.Second),
				}
				job, err := dequeue(opt)
				if err != nil {
					if !errors.Is(err, ErrEmptyQueue) {
						errFunc(err)
					}
					return err
				}
				err = handle(job, opt)
				if err != nil {
					return err
				}
				ackJobs = append(ackJobs, job)
				if len(ackJobs) >= 1000 {
					err := flush()
					if err != nil {
						errFunc(err)
						return err
					}
				}
				return nil
			}()
		}
	}
}
func getDequeueFunc(queue Queue) DequeueFunc {
	bulkDeq, ok := queue.(BulkDequeuer)
	if !ok {
		return queue.Dequeue
	}
	var jobs []*Job
	return func(opt *DequeueOptions) (*Job, error) {
		if len(jobs) == 0 {
			count := 60 / opt.InvisibleSec
			if count <= 0 {
				count = 1
			}
			bulkOpt := *opt
			bulkOpt.InvisibleSec *= count
			var err error
			jobs, err = bulkDeq.BulkDequeue(count, &bulkOpt)
			if err != nil {
				return nil, err
			}
		}
		job := jobs[0]
		jobs = jobs[1:]
		return job, nil
	}
}