func (m *Manager) Stats(ctx context.Context, request *StatsRequest) (*Stats, error) {
	return m.st.Stats(ctx, request)
}
-- Stats, Lookup and List --
Stats returns current statistics about the job queue.
package jobqueue
import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
	"github.com/google/uuid"
)
const (
	defaultConcurrency = 5
)
func nop() {}
type Manager struct {
	logger  Logger
	st      Store  
	backoff BackoffFunc
	mu               sync.Mutex            
	tm               map[string]Processor  
	concurrency      map[int]int           
	working          map[int]int           
	started          bool
	workers          map[int][]*worker
	stopSched        chan struct{}  
	workersWg        sync.WaitGroup
	jobc             map[int]chan *Job
	startupBehaviour StartupBehaviour  
	testManagerStarted   func()  
	testManagerStopped   func()  
	testSchedulerStarted func()  
	testSchedulerStopped func()  
	testJobAdded         func()  
	testJobScheduled     func()  
	testJobStarted       func()  
	testJobRetry         func()  
	testJobFailed        func()  
	testJobSucceeded     func()  
}
func New(options ...ManagerOption) *Manager {
	m := &Manager{
		logger:               stdLogger{},
		st:                   NewInMemoryStore(),
		backoff:              exponentialBackoff,
		tm:                   make(map[string]Processor),
		concurrency:          map[int]int{0: defaultConcurrency},
		working:              map[int]int{0: 0},
		startupBehaviour:     None,
		testManagerStarted:   nop,
		testManagerStopped:   nop,
		testSchedulerStarted: nop,
		testSchedulerStopped: nop,
		testJobAdded:         nop,
		testJobScheduled:     nop,
		testJobStarted:       nop,
		testJobRetry:         nop,
		testJobFailed:        nop,
		testJobSucceeded:     nop,
	}
	for _, opt := range options {
		opt(m)
	}
	return m
}
type ManagerOption func(*Manager)
func SetLogger(logger Logger) ManagerOption {
	return func(m *Manager) {
		m.logger = logger
	}
}
func SetStore(store Store) ManagerOption {
	return func(m *Manager) {
		m.st = store
	}
}
func SetBackoffFunc(fn BackoffFunc) ManagerOption {
	return func(m *Manager) {
		if fn != nil {
			m.backoff = fn
		} else {
			m.backoff = exponentialBackoff
		}
	}
}
func SetConcurrency(rank, n int) ManagerOption {
	return func(m *Manager) {
		if n <= 1 {
			m.concurrency[rank] = 1
		} else {
			m.concurrency[rank] = n
		}
	}
}
type StartupBehaviour int
const (
	None StartupBehaviour = iota
	MarkAsFailed
)
func SetStartupBehaviour(b StartupBehaviour) ManagerOption {
	return func(m *Manager) {
		m.startupBehaviour = b
	}
}
func (m *Manager) Register(topic string, p Processor) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if _, found := m.tm[topic]; found {
		return fmt.Errorf("jobqueue: topic %s already registered", topic)
	}
	m.tm[topic] = p
	return nil
}
func (m *Manager) Start() error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.started {
		return errors.New("jobqueue: manager already started")
	}
	err := m.st.Start(m.startupBehaviour)
	if err != nil {
		return err
	}
	m.jobc = make(map[int]chan *Job)
	m.workers = make(map[int][]*worker)
	for rank, concurrency := range m.concurrency {
		m.jobc[rank] = make(chan *Job, concurrency)
		m.workers[rank] = make([]*worker, concurrency)
		for i := 0; i < m.concurrency[rank]; i++ {
			m.workersWg.Add(1)
			m.workers[rank][i] = newWorker(m, m.jobc[rank])
		}
	}
	m.stopSched = make(chan struct{})
	go m.schedule()
	m.started = true
	m.testManagerStarted()  
	return nil
}
func (m *Manager) Stop() error {
	return m.Close()
}
func (m *Manager) Close() error {
	return m.CloseWithTimeout(-1 * time.Second)
}
func (m *Manager) CloseWithTimeout(timeout time.Duration) error {
	m.mu.Lock()
	if !m.started {
		m.mu.Unlock()
		return nil
	}
	m.mu.Unlock()
	m.stopSched <- struct{}{}
	<-m.stopSched
	close(m.stopSched)
	m.mu.Lock()
	for rank := range m.jobc {
		close(m.jobc[rank])
	}
	m.mu.Unlock()
	if timeout.Nanoseconds() < 0 {
		m.workersWg.Wait()
		m.testManagerStopped()  
		return nil
	}
	complete := make(chan struct{}, 1)
	go func() {
		m.workersWg.Wait()
		close(complete)
	}()
	var err error
	select {
	case <-complete:  
	case <-time.After(timeout):
		err = errors.New("jobqueue: close timed out")
	}
	m.mu.Lock()
	m.started = false
	m.mu.Unlock()
	m.testManagerStopped()  
	return err
}
func (m *Manager) Add(ctx context.Context, job *Job) error {
	if job.Topic == "" {
		return errors.New("jobqueue: no topic specified")
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	_, found := m.tm[job.Topic]
	if !found {
		return fmt.Errorf("jobqueue: topic %s not registered", job.Topic)
	}
	job.ID = uuid.New().String()
	job.State = Waiting
	job.Retry = 0
	job.Priority = -time.Now().UnixNano()
	job.Created = time.Now().UnixNano()
	err := m.st.Create(ctx, job)
	if err != nil {
		return err
	}
	m.testJobAdded()  
	return nil
}