Merge pull request #38 from enriquebris/development

v0.10.0

.travis.yml

@@ -4,6 +4,9 @@ go:
   - 1.9.x
   - 1.10.x
   - 1.11.x
+  - 1.12.x
+  - 1.13.x
+  - 1.14.x
 
 before_install:
   - go get -t -v ./...
@@ -12,4 +15,4 @@ script:
   - go test -coverprofile=coverage.txt -covermode=atomic
 
 after_success:
-  - bash <(curl -s https://codecov.io/bash)
+  - bash <(curl -s https://codecov.io/bash)

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2019 Enrique Bris
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

action.go

@@ -0,0 +1,27 @@
+// ** goworkerpool.com  **********************************************************************************************
+// ** github.com/enriquebris/goworkerpool  																			**
+// ** v0.10.0  *******************************************************************************************************
+
+package goworkerpool
+
+const (
+	actionKillWorker         = "kill.worker"
+	actionLateKillWorker     = "late.kill.worker" // late kill worker
+	actionLateKillAllWorkers = "late.kill.all.workers"
+	actionLateKillWorkers    = "late.kill.workers"
+	actionPauseAllWorkers    = "pause.all.workers"
+)
+
+// action represents an action
+// An action would be normally sent to a worker, unless SendToWorker was false.
+// The PreExternalFunc function (if any) would be executed by the dispatcher just before send the action to a worker. In
+// case that SendToWorker was false, the dispatcher would execute the function and continue to the next iteration.
+type action struct {
+	// action code
+	Code string
+	// to let the dispatcher knows whether the action should be sent to a worker. The dispatcher would process actions
+	// if they are not sent to workers.
+	SendToWorker bool
+	// to be executed by the dispatcher just before send the action to a worker (in case the action was to be sent to the worker)
+	PreExternalFunc dispatcherGeneralFunc
+}

error.go

@@ -0,0 +1,52 @@
+// ** goworkerpool.com  **********************************************************************************************
+// ** github.com/enriquebris/goworkerpool  																			**
+// ** v0.10.0  *******************************************************************************************************
+
+package goworkerpool
+
+const (
+	ErrorEmbeddedErrors                         = "embedded.errors"
+	ErrorData                                   = "wrong.data"
+	ErrorKillAllWorkersInProgress               = "action.kill.all.workers.in.progress"
+	ErrorWaitInProgress                         = "action.wait.in.progress"
+	ErrorWaitUntilInitialWorkersAreUpInProgress = "action.wait.until.initial.workers.are.up.in.progress"
+	ErrorDispatcherNoActionNoTask               = "dispatcher.no.action.no.task"
+	ErrorDispatcherWorkerCouldNotBeEnqueued     = "dispatcher.worker.couldnt.be.enqueued"
+	ErrorNoDefaultWorkerFunction                = "no.default.worker.function"
+	ErrorDispatcherChannelFull                  = "dispatcher.channel.at.full.capacity"
+	ErrorFullCapacityChannel                    = "full.capacity.channel"
+	ErrorWorkerNotFound                         = "worker.not.found"
+	ErrorWorkerType                             = "worker.wrong.type"
+	ErrorWorkerMaxTimeExceeded                  = "worker.max.time.exceeded"
+	ErrorWorkerFuncPanic                        = "worker.func.panic"
+)
+
+type PoolError struct {
+	code           string
+	message        string
+	embeddedErrors []error
+}
+
+func newPoolError(code string, message string) *PoolError {
+	return &PoolError{
+		code:    code,
+		message: message,
+	}
+}
+
+func (st *PoolError) Error() string {
+	return st.message
+}
+
+func (st *PoolError) Code() string {
+	return st.code
+}
+
+// AddEmbeddedErrors adds a slice of errors
+func (st *PoolError) AddEmbeddedErrors(errors []error) {
+	st.embeddedErrors = errors
+}
+
+func (st *PoolError) GetEmbeddedErrors() []error {
+	return st.embeddedErrors
+}

error_test.go

@@ -0,0 +1,35 @@
+// ** goworkerpool.com  **********************************************************************************************
+// ** github.com/enriquebris/goworkerpool  																			**
+// ** v0.10.0  *******************************************************************************************************
+
+package goworkerpool
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type PoolErrorTestSuite struct {
+	suite.Suite
+}
+
+// ***************************************************************************************
+// ** Initialization
+// ***************************************************************************************
+
+// no elements at initialization
+func (suite *PoolErrorTestSuite) TestInitialization() {
+	queueError := newPoolError("code", "message")
+
+	suite.Equal("code", queueError.Code())
+	suite.Equal("message", queueError.Error())
+}
+
+// ***************************************************************************************
+// ** Run suite
+// ***************************************************************************************
+
+func TestQueueErrorTestSuite(t *testing.T) {
+	suite.Run(t, new(PoolErrorTestSuite))
+}

examples/adjust_total_workers_on_demand/main.go

@@ -6,21 +6,34 @@
 package main
 
 import (
+	"fmt"
 	"log"
 	"time"
 
 	"github.com/enriquebris/goworkerpool"
 )
 
 func main() {
-	// total workers
-	totalWorkers := 10
-	// max number of pending jobs
-	maxNumberPendingJobs := 15
-	// do not show messages about the pool processing
-	verbose := false
+	var (
+		// total workers
+		totalWorkers uint = 10
+		// max number of pending jobs
+		maxNumberPendingJobs uint = 150
+		// do not show messages about the pool processing
+		verbose = false
+	)
 
-	pool := goworkerpool.NewPool(totalWorkers, maxNumberPendingJobs, verbose)
+	pool, err := goworkerpool.NewPoolWithOptions(goworkerpool.PoolOptions{
+		TotalInitialWorkers:  totalWorkers,
+		MaxOperationsInQueue: maxNumberPendingJobs,
+		MaxWorkers:           20,
+		LogVerbose:           verbose,
+	})
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
 
 	// add the worker handler function
 	pool.SetWorkerFunc(func(data interface{}) bool {
@@ -34,7 +47,7 @@ func main() {
 	})
 
 	// start up the workers and wait until them are up
-	pool.StartWorkersAndWait()
+	pool.WaitUntilInitialWorkersAreUp()
 
 	// enqueue jobs in a separate goroutine
 	go func() {

examples/basic/main.go

@@ -6,21 +6,29 @@
 package main
 
 import (
+	"fmt"
 	"log"
 	"time"
 
 	"github.com/enriquebris/goworkerpool"
 )
 
 func main() {
-	// total workers
-	totalWorkers := 10
-	// max number of pending jobs
-	maxNumberPendingJobs := 15
-	// do not log messages about the pool processing
-	verbose := false
+	var (
+		maxOperationsInQueue uint = 50
+	)
+	pool, err := goworkerpool.NewPoolWithOptions(goworkerpool.PoolOptions{
+		TotalInitialWorkers:          10,
+		MaxWorkers:                   20,
+		MaxOperationsInQueue:         maxOperationsInQueue,
+		WaitUntilInitialWorkersAreUp: true,
+		LogVerbose:                   true,
+	})
 
-	pool := goworkerpool.NewPool(totalWorkers, maxNumberPendingJobs, verbose)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
 
 	// add the worker handler function
 	pool.SetWorkerFunc(func(data interface{}) bool {
@@ -33,12 +41,9 @@ func main() {
 		return true
 	})
 
-	// start up the workers and wait until them are up
-	pool.StartWorkersAndWait()
-
 	// enqueue jobs in a separate goroutine
 	go func() {
-		for i := 0; i < 30; i++ {
+		for i := 0; i < int(maxOperationsInQueue); i++ {
 			pool.AddTask(i)
 		}
 

examples/callback/main.go

@@ -7,21 +7,34 @@
 package main
 
 import (
+	"fmt"
 	"log"
 	"time"
 
 	"github.com/enriquebris/goworkerpool"
 )
 
 func main() {
-	// total workers
-	totalWorkers := 10
-	// max number of pending jobs
-	maxNumberPendingJobs := 15
-	// do not log messages about the pool processing
-	verbose := false
+	var (
+		// total workers
+		totalWorkers uint = 10
+		// max number of pending jobs
+		maxNumberPendingJobs uint = 150
+		// do not show messages about the pool processing
+		verbose = false
+	)
 
-	pool := goworkerpool.NewPool(totalWorkers, maxNumberPendingJobs, verbose)
+	pool, err := goworkerpool.NewPoolWithOptions(goworkerpool.PoolOptions{
+		TotalInitialWorkers:  totalWorkers,
+		MaxOperationsInQueue: maxNumberPendingJobs,
+		MaxWorkers:           20,
+		LogVerbose:           verbose,
+	})
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
 
 	// add the worker handler function
 	pool.SetWorkerFunc(func(data interface{}) bool {
@@ -35,7 +48,7 @@ func main() {
 	})
 
 	// start up the workers and wait until them are up
-	pool.StartWorkersAndWait()
+	pool.WaitUntilInitialWorkersAreUp()
 
 	// enqueue jobs in a separate goroutine
 	go func() {

examples/complex_jobs/main.go

@@ -19,21 +19,31 @@ type JobData struct {
 }
 
 func main() {
-	// total workers
-	totalWorkers := 10
-	// max number of pending jobs
-	maxNumberPendingJobs := 15
-	// do not show messages about the pool processing
-	verbose := false
+	var (
+		// total workers
+		totalWorkers uint = 10
+		// max number of pending jobs
+		maxNumberPendingJobs uint = 150
+		// do not show messages about the pool processing
+		verbose = false
+	)
 
-	pool := goworkerpool.NewPool(totalWorkers, maxNumberPendingJobs, verbose)
+	pool, err := goworkerpool.NewPoolWithOptions(goworkerpool.PoolOptions{
+		TotalInitialWorkers:          totalWorkers,
+		MaxOperationsInQueue:         maxNumberPendingJobs,
+		MaxWorkers:                   20,
+		WaitUntilInitialWorkersAreUp: true,
+		LogVerbose:                   verbose,
+	})
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
 
 	// add the worker handler function
 	pool.SetWorkerFunc(worker)
 
-	// start up the workers and wait until them are up
-	pool.StartWorkersAndWait()
-
 	// enqueue jobs in a separate goroutine
 	go func() {
 		for i := 0; i < 30; i++ {