func (st *Pool) getNoWorkersChan() chan struct{} {
	return st.noWorkersChan
}
getNoWorkersChan returns the channel over which the "no workers" signal travels
package goworkerpool
import (
	"fmt"
	"sync"
	"github.com/enriquebris/goconcurrentcounter"
	"github.com/enriquebris/goconcurrentqueue"
)
const (
	waitInProgress = "wait.in.progress"
	killAllWorkersInProgress = "kill.all.workers.in.progress"
	dispatcherInProgress = "dispatcher.in.progress"
	triggerZeroTotalWorkers    = "trigger.zero.total.workers"
	triggerWaitUntilNSuccesses = "trigger.wait.until.n.successes"
	defaultMaxWorkers = 100
)
type PoolFunc func(interface{}) bool
type PoolCallback func(interface{})
type dispatcherGeneralFunc func()
type Pool struct {
	tasks goconcurrentqueue.Queue
	panicTasks goconcurrentqueue.Queue
	actions goconcurrentqueue.Queue
	workers sync.Map
	availableWorkers goconcurrentqueue.Queue
	defaultWorkerFunc PoolFunc
	activityChan chan struct{}
	dispatcherWorkerFeedbackChan chan workerFeedback
	resumeWorkers chan struct{}
	noWorkersChan chan struct{}
	waitUntilInitialWorkersAreUp chan struct{}
	newWorkerID goconcurrentcounter.Int
	totalWorkers goconcurrentcounter.Int
	taskSuccesses goconcurrentcounter.Int
	totalWorkersInProgress goconcurrentcounter.Int
	externalNewWorkerChan chan int
	externalKilledWorkerChan chan int
	logVerbose bool
	logChan    chan PoolLog
	generalInfoMap sync.Map
}
func NewPool(initialWorkers int, maxOperationsInQueue int, logVerbose bool) *Pool {
	if initialWorkers < 0 || maxOperationsInQueue <= 0 {
		return nil
	}
	ret := &Pool{}
	ret.initialize(PoolOptions{
		TotalInitialWorkers:          uint(initialWorkers),
		MaxWorkers:                   defaultMaxWorkers,
		MaxOperationsInQueue:         uint(maxOperationsInQueue),
		WaitUntilInitialWorkersAreUp: false,
		LogVerbose:                   logVerbose,
	})
	return ret
}
func NewPoolWithOptions(options PoolOptions) (*Pool, error) {
	if options.MaxWorkers == 0 {
		return nil, newPoolError(ErrorData, "MaxWorkers should be greater than zero")
	}
	if options.MaxOperationsInQueue == 0 {
		return nil, newPoolError(ErrorData, "MaxOperationsInQueue should be greater than zero")
	}
	pool := &Pool{}
	pool.SetNewWorkerChan(options.NewWorkerChan)
	pool.initialize(options)
	if options.WaitUntilInitialWorkersAreUp {
		<-pool.waitUntilInitialWorkersAreUp
	}
	return pool, nil
}
func (st *Pool) initialize(options PoolOptions) {
	st.tasks = goconcurrentqueue.NewFIFO()
	st.panicTasks = goconcurrentqueue.NewFIFO()
	st.actions = goconcurrentqueue.NewFIFO()
	st.availableWorkers = goconcurrentqueue.NewFIFO()
	st.activityChan = make(chan struct{}, int(options.MaxOperationsInQueue))
	st.dispatcherWorkerFeedbackChan = make(chan workerFeedback, int(options.MaxWorkers))
	st.noWorkersChan = make(chan struct{})
	st.waitUntilInitialWorkersAreUp = make(chan struct{}, 2)
	st.newWorkerID = goconcurrentcounter.NewIntChan(0)
	st.totalWorkers = goconcurrentcounter.NewIntChan(0)
	st.totalWorkers.SetTriggerOnValue(int(options.TotalInitialWorkers), "WaitUntilInitialWorkersAreUp", func() {
		st.totalWorkers.EnqueueToRunAfterCurrentTriggerFunctions(func() {
			st.totalWorkers.UnsetTriggerOnValue(int(options.TotalInitialWorkers), "WaitUntilInitialWorkersAreUp")
		})
		st.waitUntilInitialWorkersAreUp <- struct{}{}
		if options.WaitUntilInitialWorkersAreUp {
			st.waitUntilInitialWorkersAreUp <- struct{}{}
		}
		close(st.waitUntilInitialWorkersAreUp)
	})
	st.taskSuccesses = goconcurrentcounter.NewIntChan(0)
	st.totalWorkersInProgress = goconcurrentcounter.NewIntChan(0)
	go st.dispatcher()
	go st.dispatcherWorkerFeedback()
	for i := 0; i < int(options.TotalInitialWorkers); i++ {
		if err := st.AddWorker(); err != nil {
			st.logError("initialize.AddWorker", err)
		}
	}
	st.totalWorkers.SetTriggerOnValue(0, triggerZeroTotalWorkers, func() {
		if st.getStatus(killAllWorkersInProgress) {
			st.setStatus(killAllWorkersInProgress, false)
		}
		if st.getStatus(waitInProgress) {
			st.getNoWorkersChan() <- struct{}{}
		}
	})
}
func (st *Pool) dispatcher() {
	keepWorking := true
	for keepWorking {
		var (
			noAction     = true
			noTask       = true
			actn         action
			tsk          *task
			preFunc      dispatcherGeneralFunc = nil
			sendToWorker bool                  = true
		)
		st.setStatus(dispatcherInProgress, true)
		select {
		case _, ok := <-st.activityChan:
			if !ok {
				keepWorking = false
				break
			}
			if rawAction, err := st.actions.Dequeue(); err != nil {
				queueError, ok := err.(*goconcurrentqueue.QueueError)
				if ok && queueError.Code() == goconcurrentqueue.QueueErrorCodeEmptyQueue {
					noAction = true
				} else {
					st.logError("dispatcher.action.dequeue", err)
					continue
				}
			} else {
				noAction = false
				actn, _ = rawAction.(action)
				preFunc = actn.PreExternalFunc
				sendToWorker = actn.SendToWorker
			}
			if noAction {
				if rawTask, err := st.tasks.Dequeue(); err != nil {
					queueError, ok := err.(*goconcurrentqueue.QueueError)
					if ok && queueError.Code() == goconcurrentqueue.QueueErrorCodeEmptyQueue {
						noTask = true
					} else {
						st.logError("dispatcher.task.dequeue", err)
						continue
					}
				} else {
					noTask = false
					tsk, _ = rawTask.(*task)
					if tsk.Code == taskLateAction {
						tmpAction, ok := tsk.Data.(action)
						if !ok {
							st.logError("dispatcher.taskLateAction unexpected type", nil)
							continue
						}
						sendToWorker = tmpAction.SendToWorker
						preFunc = tmpAction.PreExternalFunc
					}
				}
			}
			if noAction && noTask {
				st.logError("no action/task at dispatcher.dequeue", newPoolError(ErrorDispatcherNoActionNoTask, "no action/task at dispatcher.dequeue"))
				continue
			} else {
				if preFunc != nil {
					preFunc()
				}
				if !sendToWorker {
					continue
				}
				if rawWorker, err := st.availableWorkers.DequeueOrWaitForNextElement(); err != nil {
					st.logError("dispatcher.availableWorker.dequeue", err)
					if noAction {
						if err := st.AddTask(tsk); err != nil {
							st.logError("dispatcher.reEnqueue.AddTask", err)
						}
					} else {
						if err := st.addAction(actn.Code, actn.SendToWorker, actn.PreExternalFunc); err != nil {
							st.logError("dispatcher.reEnqueue.AddAction", err)
						}
					}
					continue
				} else {
					wkr, _ := rawWorker.(*worker)
					if noTask {
						wkr.ProcessAction(actn)
					} else {
						wkr.ProcessTask(tsk)
					}
				}
			}
		}
	}
	st.setStatus(dispatcherInProgress, false)
}
func (st *Pool) dispatcherWorkerFeedback() {
	keepWorking := true
	for keepWorking {
		select {
		case workerFeedback := <-st.dispatcherWorkerFeedbackChan:
			rawWorker, ok := st.workers.Load(workerFeedback.workerID)
			if !ok {
				st.logError("dispatcherWorkerFeedback.unknown worker", newPoolError(ErrorWorkerNotFound, fmt.Sprintf("Missing worker %v could not be removed", workerFeedback.workerID)))
				continue
			}
			wkr, ok := rawWorker.(*worker)
			if !ok {
				st.logError("dispatcherWorkerFeedback.worker", newPoolError(ErrorWorkerType, fmt.Sprintf("Wrong worker's type for id: %v", workerFeedback.workerID)))
				continue
			}
			if workerFeedback.reEnqueueWorker {
				if err := st.availableWorkers.Enqueue(wkr); err != nil {
					st.logError("dispatcherWorkerFeedback.availableWorkers.Enqueue", err)
				} else {
					continue
				}
			}
			if err := st.removeWorker(wkr.GetID()); err != nil {
				st.logError("dispatcherWorkerFeedback.removeWorker", err)
			}
		}
	}
}
func (st *Pool) addTask(taskType int, data interface{}, fn PoolFunc, callbackFn PoolCallback) error {
	job := &task{
		Code:         taskType,
		Data:         data,
		Func:         fn,
		CallbackFunc: callbackFn,
		Valid:        true,
	}
	if err := st.tasks.Enqueue(job); err != nil {
		return err
	}
	select {
	case st.activityChan <- struct{}{}:
	default:
		job.Valid = false
		return newPoolError(ErrorDispatcherChannelFull, "task couldn't be enqueued because the operations(tasks/actions) queue is at full capacity")
	}
	return nil
}
func (st *Pool) addAction(actionCode string, sendToWorker bool, preExternalFunc dispatcherGeneralFunc) error {
	actn := action{
		Code:            actionCode,
		SendToWorker:    sendToWorker,
		PreExternalFunc: preExternalFunc,
	}
	if err := st.actions.Enqueue(actn); err != nil {
		return err
	}
	select {
	case st.activityChan <- struct{}{}:
	default:
		return newPoolError(ErrorDispatcherChannelFull, "action couldn't be enqueued because the operations(tasks/actions) queue is at full capacity")
	}
	return nil
}
func (st *Pool) AddTask(data interface{}) error {
	if st.getStatus(killAllWorkersInProgress) {
		return newPoolError(ErrorKillAllWorkersInProgress, messageKillAllWorkersInProgress)
	}
	if st.defaultWorkerFunc == nil {
		return newPoolError(ErrorNoDefaultWorkerFunction, "Missing default handler function to process this task")
	}
	return st.addTask(taskRegular, data, st.defaultWorkerFunc, nil)
}
func (st *Pool) AddTaskCallback(data interface{}, callbackFn PoolCallback) error {
	if callbackFn == nil {
		return newPoolError(ErrorData, "Callback function should not be nil")
	}
	if st.getStatus(killAllWorkersInProgress) {
		return newPoolError(ErrorKillAllWorkersInProgress, messageKillAllWorkersInProgress)
	}
	if st.defaultWorkerFunc == nil {
		return newPoolError(ErrorNoDefaultWorkerFunction, "Missing default handler function to process this task")
	}
	return st.addTask(taskRegular, data, st.defaultWorkerFunc, callbackFn)
}
func (st *Pool) AddCallback(callbackFn PoolCallback) error {
	if callbackFn == nil {
		return newPoolError(ErrorData, "Callback function should not be nil")
	}
	if st.getStatus(killAllWorkersInProgress) {
		return newPoolError(ErrorKillAllWorkersInProgress, messageKillAllWorkersInProgress)
	}
	if st.defaultWorkerFunc == nil {
		return newPoolError(ErrorNoDefaultWorkerFunction, "Missing default handler function to process this task")
	}
	return st.addTask(taskRegular, nil, nil, callbackFn)
}
func (st *Pool) StartWorkers() error {
	return nil
}
func (st *Pool) StartWorkersAndWait() error {
	return st.WaitUntilInitialWorkersAreUp()
}
func (st *Pool) SetWorkerFunc(fn PoolFunc) {
	st.defaultWorkerFunc = fn
}