func (w *workManager) Work(j job) queuedJob {
	qj := newQueuedJob(j)
	switch j.Type() {
	case collectJobType:
		w.collectq.Event <- qj
	case processJobType:
		w.processq.Event <- qj
	case publishJobType:
		w.publishq.Event <- qj
	}
	return qj
}
Work dispatches jobs to worker pools for processing.

Returns a queued job to the caller, which will be
completed by the work queue aubsystem.
package scheduler
import "sync"
type workManager struct {
	state          workManagerState
	collectq       *queue
	publishq       *queue
	processq       *queue
	collectWkrs    []*worker
	publishWkrs    []*worker
	processWkrs    []*worker
	collectQSize   uint
	publishQSize   uint
	processQSize   uint
	collectWkrSize uint
	publishWkrSize uint
	processWkrSize uint
	collectchan    chan queuedJob
	publishchan    chan queuedJob
	processchan    chan queuedJob
	kill           chan struct{}
	mutex          *sync.Mutex
}
type workManagerState int
const (
	workManagerStopped workManagerState = iota
	workManagerRunning
	defaultQSize   uint = 5
	defaultWkrSize uint = 1
)
type workManagerOption func(w *workManager) workManagerOption
func CollectQSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.collectQSize
		w.collectQSize = v
		return CollectQSizeOption(previous)
	}
}
func PublishQSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.publishQSize
		w.publishQSize = v
		return PublishQSizeOption(previous)
	}
}
func ProcessQSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.processQSize
		w.processQSize = v
		return ProcessQSizeOption(previous)
	}
}
func CollectWkrSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.collectWkrSize
		w.collectWkrSize = v
		return CollectWkrSizeOption(previous)
	}
}
func ProcessWkrSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.processWkrSize
		w.processWkrSize = v
		return ProcessWkrSizeOption(previous)
	}
}
func PublishWkrSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.publishWkrSize
		w.publishWkrSize = v
		return PublishWkrSizeOption(previous)
	}
}
func newWorkManager(opts ...workManagerOption) *workManager {
	wm := &workManager{
		collectQSize:   defaultQSize,
		processQSize:   defaultQSize,
		publishQSize:   defaultQSize,
		collectWkrSize: defaultWkrSize,
		publishWkrSize: defaultWkrSize,
		processWkrSize: defaultWkrSize,
		collectchan:    make(chan queuedJob),
		publishchan:    make(chan queuedJob),
		processchan:    make(chan queuedJob),
		kill:           make(chan struct{}),
		mutex:          &sync.Mutex{},
	}
	for _, opt := range opts {
		opt(wm)
	}
	wm.collectq = newQueue(wm.collectQSize, wm.sendToWorker)
	wm.publishq = newQueue(wm.publishQSize, wm.sendToWorker)
	wm.processq = newQueue(wm.processQSize, wm.sendToWorker)
	wm.publishq.Start()
	wm.collectq.Start()
	wm.processq.Start()
	wm.collectWkrs = make([]*worker, wm.collectWkrSize)
	var i uint
	for i = 0; i < wm.collectWkrSize; i++ {
		wm.collectWkrs[i] = newWorker(wm.collectchan)
		go wm.collectWkrs[i].start()
	}
	wm.publishWkrs = make([]*worker, wm.publishWkrSize)
	for i = 0; i < wm.publishWkrSize; i++ {
		wm.publishWkrs[i] = newWorker(wm.publishchan)
		go wm.publishWkrs[i].start()
	}
	wm.processWkrs = make([]*worker, wm.processWkrSize)
	for i = 0; i < wm.processWkrSize; i++ {
		wm.processWkrs[i] = newWorker(wm.processchan)
		go wm.processWkrs[i].start()
	}
	return wm
}
func (w *workManager) Start() {
	w.mutex.Lock()
	defer w.mutex.Unlock()
	if w.state == workManagerStopped {
		w.state = workManagerRunning
		go func() {
			for {
				select {
				case <-w.collectq.Err:
				case <-w.processq.Err:
				case <-w.publishq.Err:
				case <-w.kill:
					return
				}
			}
		}()
	}
}
func (w *workManager) Stop() {
	w.collectq.Stop()
	close(workerKillChan)
	close(w.kill)
}