Merge pull request #199 from RichardKnop/feature/add-async-worker-launch-method

Add a non blocking LaunchAsync method to worker

Author: RichardKnop

example/machinery.go

@@ -4,8 +4,6 @@ import (
 	"errors"
 	"fmt"
 	"os"
-	"reflect"
-	"strings"
 	"time"
 
 	"github.com/RichardKnop/machinery/v1"
@@ -197,7 +195,7 @@ func send() error {
 	if err != nil {
 		return fmt.Errorf("Getting task result failed with error: %s", err.Error())
 	}
-	log.INFO.Printf("1 + 1 = %v\n", humanReadableResults(results))
+	log.INFO.Printf("1 + 1 = %v\n", tasks.HumanReadableResults(results))
 
 	/*
 	 * Now let's explore ways of sending multiple tasks
@@ -222,7 +220,7 @@ func send() error {
 			"%v + %v = %v\n",
 			asyncResult.Signature.Args[0].Value,
 			asyncResult.Signature.Args[1].Value,
-			humanReadableResults(results),
+			tasks.HumanReadableResults(results),
 		)
 	}
 
@@ -241,7 +239,7 @@ func send() error {
 	if err != nil {
 		return fmt.Errorf("Getting chord result failed with error: %s", err.Error())
 	}
-	log.INFO.Printf("(1 + 1) * (2 + 2) * (5 + 6) = %v\n", humanReadableResults(results))
+	log.INFO.Printf("(1 + 1) * (2 + 2) * (5 + 6) = %v\n", tasks.HumanReadableResults(results))
 
 	// Now let's try chaining task results
 	initTasks()
@@ -257,7 +255,7 @@ func send() error {
 	if err != nil {
 		return fmt.Errorf("Getting chain result failed with error: %s", err.Error())
 	}
-	log.INFO.Printf("(((1 + 1) + (2 + 2)) + (5 + 6)) * 4 = %v\n", humanReadableResults(results))
+	log.INFO.Printf("(((1 + 1) + (2 + 2)) + (5 + 6)) * 4 = %v\n", tasks.HumanReadableResults(results))
 
 	// Let's try a task which throws panic to make sure stack trace is not lost
 	initTasks()
@@ -283,20 +281,7 @@ func send() error {
 	if err != nil {
 		return fmt.Errorf("Getting long running task result failed with error: %s", err.Error())
 	}
-	log.INFO.Printf("Long running task returned = %v\n", humanReadableResults(results))
+	log.INFO.Printf("Long running task returned = %v\n", tasks.HumanReadableResults(results))
 
 	return nil
 }
-
-func humanReadableResults(results []reflect.Value) string {
-	if len(results) == 1 {
-		return fmt.Sprintf("%v", results[0].Interface())
-	}
-
-	readableResults := make([]string, len(results))
-	for i := 0; i < len(results); i++ {
-		readableResults[i] = fmt.Sprintf("%v", results[i].Interface())
-	}
-
-	return fmt.Sprintf("[%s]", strings.Join(readableResults, ", "))
-}

v1/backends/async_result.go

@@ -87,15 +87,7 @@ func (asyncResult *AsyncResult) Touch() ([]reflect.Value, error) {
 	}
 
 	if asyncResult.taskState.IsSuccess() {
-		resultValues := make([]reflect.Value, len(asyncResult.taskState.Results))
-		for i, result := range asyncResult.taskState.Results {
-			resultValue, err := tasks.ReflectValue(result.Type, result.Value)
-			if err != nil {
-				return nil, err
-			}
-			resultValues[i] = resultValue
-		}
-		return resultValues, nil
+		return tasks.ReflectTaskResults(asyncResult.taskState.Results)
 	}
 
 	return nil, nil

v1/tasks/result.go

@@ -1,7 +1,40 @@
 package tasks
 
+import (
+	"fmt"
+	"reflect"
+	"strings"
+)
+
 // TaskResult represents an actual return value of a processed task
 type TaskResult struct {
 	Type  string      `bson:"type"`
 	Value interface{} `bson:"value"`
 }
+
+// ReflectTaskResults ...
+func ReflectTaskResults(taskResults []*TaskResult) ([]reflect.Value, error) {
+	resultValues := make([]reflect.Value, len(taskResults))
+	for i, taskResult := range taskResults {
+		resultValue, err := ReflectValue(taskResult.Type, taskResult.Value)
+		if err != nil {
+			return nil, err
+		}
+		resultValues[i] = resultValue
+	}
+	return resultValues, nil
+}
+
+// HumanReadableResults ...
+func HumanReadableResults(results []reflect.Value) string {
+	if len(results) == 1 {
+		return fmt.Sprintf("%v", results[0].Interface())
+	}
+
+	readableResults := make([]string, len(results))
+	for i := 0; i < len(results); i++ {
+		readableResults[i] = fmt.Sprintf("%v", results[i].Interface())
+	}
+
+	return fmt.Sprintf("[%s]", strings.Join(readableResults, ", "))
+}

v1/worker.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
-	"strings"
 	"syscall"
 	"time"
 
@@ -25,9 +24,19 @@ type Worker struct {
 // Launch starts a new worker process. The worker subscribes
 // to the default queue and processes incoming registered tasks
 func (worker *Worker) Launch() error {
+	errorsChan := make(chan error)
+
+	worker.LaunchAsync(errorsChan)
+
+	return <-errorsChan
+}
+
+// LaunchAsync is a non blocking version of Launch
+func (worker *Worker) LaunchAsync(errorsChan chan<- error) {
 	cnf := worker.server.GetConfig()
 	broker := worker.server.GetBroker()
 
+	// Log some useful information about woorker configuration
 	log.INFO.Printf("Launching a worker with the following settings:")
 	log.INFO.Printf("- Broker: %s", cnf.Broker)
 	log.INFO.Printf("- DefaultQueue: %s", cnf.DefaultQueue)
@@ -40,11 +49,7 @@ func (worker *Worker) Launch() error {
 		log.INFO.Printf("  - PrefetchCount: %d", cnf.AMQP.PrefetchCount)
 	}
 
-	errorsChan := make(chan error)
-	sig := make(chan os.Signal, 1)
-	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
-	var signalsReceived uint
-
+	// Goroutine to start broker consumption and handle retries when broker connection dies
 	go func() {
 		for {
 			retry, err := broker.StartConsuming(worker.ConsumerTag, worker.Concurrency, worker)
@@ -58,6 +63,11 @@ func (worker *Worker) Launch() error {
 		}
 	}()
 
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+	var signalsReceived uint
+
+	// Goroutine Handle SIGINT and SIGTERM signals
 	go func() {
 		for {
 			select {
@@ -79,8 +89,6 @@ func (worker *Worker) Launch() error {
 			}
 		}
 	}()
-
-	return <-errorsChan
 }
 
 // Quit tears down the running worker process
@@ -166,11 +174,15 @@ func (worker *Worker) taskSucceeded(signature *tasks.Signature, taskResults []*t
 		return fmt.Errorf("Set state success error: %s", err)
 	}
 
-	debugResults := make([]string, len(taskResults))
-	for i, taskResult := range taskResults {
-		debugResults[i] = fmt.Sprintf("%v", taskResult.Value)
+	// Log human readable results of the processed task
+	var debugResults = "[]"
+	results, err := tasks.ReflectTaskResults(taskResults)
+	if err != nil {
+		log.WARNING.Print(err)
+	} else {
+		debugResults = tasks.HumanReadableResults(results)
 	}
-	log.INFO.Printf("Processed task %s. Results = [%v]", signature.UUID, strings.Join(debugResults, ", "))
+	log.INFO.Printf("Processed task %s. Results = %s", signature.UUID, debugResults)
 
 	// Trigger success callbacks