change return from a task to []any

Author: manhvu

README.md

@@ -56,7 +56,7 @@ import "github.com/manhvu/easyworker"
 
 ## Use Cases
 
-### Supervisor
+### Supervisor model
 
 This is used for generic worker(child).
 Every children has a owner restart strategy.
@@ -71,6 +71,9 @@ Children will be started after they are added to supervisor.
 
 In restart case, children will re-use last parameters (if task don't change it) of task.
 
+Child, doesn't return any value from task.
+You need add code to get value from task if you needed.
+
 Supervisor -> Child -> call user functions
 
 Basic supervisor's flow:
@@ -178,6 +181,9 @@ Number of workers is number of goroutine will run tasks.
 
 In retry case, worker will re-use last parameters of task.
 
+Result of each task is a []any.
+You need to get true value from any(interface{}).
+
 EasyTask example:
 
 ```go
@@ -230,6 +236,9 @@ Number of workers is number of goroutines used for processing streaming task.
 
 In retry case, workers will re-use last parameters of task.
 
+Result of each task is a []any.
+You need to get true value from any(interface{}).
+
 EasyStream example:
 
 ```go
@@ -258,19 +267,19 @@ myStream.Run()
 
 // receive data from stream.
 go func() {
-    for {
-        r := <-outCh
-        fmt.Println("stream result: ", r)
-    }
+  for {
+    r := <-outCh
+    fmt.Println("stream result: ", r)
+  }
 }()
 
 // send data to stream.
 go func() {
-    for i := 0; i < 15; i++ {
-        input := []any{i, "hello"}
-        inCh <- input
-        fmt.Println("stream sent: ", input)
-    }
+  for i := 0; i < 15; i++ {
+    input := []any{i, "hello"}
+    inCh <- input
+    fmt.Println("stream sent: ", input)
+  }
 }()
 
 
@@ -290,6 +299,8 @@ Second param is channel that user can receive signal.
 
 Signal is a struct with reference id and kind of end (failed, done).
 
+Go doesn't support return value, you need an other way to get if you neened.
+
 Example 1:
 
 ```go

caller.go

@@ -9,7 +9,7 @@ import (
 /*
 call user's function througth reflect.
 */
-func invokeFun(fun any, args ...any) (ret reflect.Value, err error) {
+func invokeFun(fun any, args ...any) (ret []any, err error) {
 	// catch if panic by user code.
 	defer func() {
 		if r := recover(); r != nil {
@@ -24,10 +24,10 @@ func invokeFun(fun any, args ...any) (ret reflect.Value, err error) {
 	fnType := fn.Type()
 	numIn := fnType.NumIn()
 	if numIn > len(args) {
-		return reflect.ValueOf(nil), fmt.Errorf("function must have minimum %d params. Have %d", numIn, len(args))
+		return nil, fmt.Errorf("function must have minimum %d params. Have %d", numIn, len(args))
 	}
 	if numIn != len(args) && !fnType.IsVariadic() {
-		return reflect.ValueOf(nil), fmt.Errorf("func must have %d params. Have %d", numIn, len(args))
+		return nil, fmt.Errorf("func must have %d params. Have %d", numIn, len(args))
 	}
 	params := make([]reflect.Value, len(args))
 	for i := 0; i < len(args); i++ {
@@ -39,24 +39,26 @@ func invokeFun(fun any, args ...any) (ret reflect.Value, err error) {
 		}
 		argValue := reflect.ValueOf(args[i])
 		if !argValue.IsValid() {
-			return reflect.ValueOf(nil), fmt.Errorf("func Param[%d] must be %s. Have %s", i, inType, argValue.String())
+			return nil, fmt.Errorf("func Param[%d] must be %s. Have %s", i, inType, argValue.String())
 		}
 		argType := argValue.Type()
 		if argType.ConvertibleTo(inType) {
 			params[i] = argValue.Convert(inType)
 		} else {
-			return reflect.ValueOf(nil), fmt.Errorf("method Param[%d] must be %s. Have %s", i, inType, argType)
+			return nil, fmt.Errorf("method Param[%d] must be %s. Have %s", i, inType, argType)
 		}
 	}
 
 	result := fn.Call(params)
 
-	//log.Println("invoke result:", result)
+	ret = make([]any, len(result))
 
-	if len(result) > 0 {
-		ret = result[0]
+	for i, r := range result {
+		ret[i] = r.Interface()
 	}
 
+	//log.Println("invoke result:", result)
+
 	return ret, nil
 }
 

caller_test.go

@@ -0,0 +1,67 @@
+package easyworker
+
+import (
+	"errors"
+	"log"
+	"testing"
+)
+
+func simpleLoopReturn(a int) int {
+	ret := 0
+	for i := 0; i < a; i++ {
+		ret += i
+	}
+
+	return ret
+}
+
+func returnMultiValue() (i int, f float64, b bool, e error) {
+	i = 123
+	f = 1.2
+	b = true
+	e = errors.New("test return")
+	return
+}
+
+func TestInvokeNoArg(t *testing.T) {
+	result, err := invokeFun(simpleLoopNoArg)
+
+	if err != nil {
+		t.Error("test invoke with no argument failed, ", err)
+	} else {
+		log.Println("result: ", result)
+	}
+}
+
+func TestInvokePanic(t *testing.T) {
+	_, err := invokeFun(simpleLoopWithPanic, 5)
+
+	if err == nil {
+		t.Error("expected error but no return error.")
+	} else {
+		log.Println("expected is ok, err: ", err)
+	}
+}
+
+func TestInvokeReturn(t *testing.T) {
+	result, err := invokeFun(simpleLoopReturn, 5)
+
+	if err != nil {
+		t.Error("test invoke with no argument failed, ", err)
+	} else {
+		log.Println("result: ", result)
+		if result[0] != 10 {
+			t.Error("return incorrect value")
+		}
+	}
+}
+
+func TestInvokeReturnMultiValue(t *testing.T) {
+	result, err := invokeFun(returnMultiValue)
+
+	if err != nil {
+		t.Error("test invoke with no argument failed, ", err)
+	} else {
+		log.Println("result: ", result)
+	}
+}

gomonitor.go

@@ -19,13 +19,14 @@ const (
 type GoSignal struct {
 	// Monitor reference id.
 	RefId int64
+
 	// Kind of signal (SIGNAL_DONE, SIGNAL_FAILED)
 	Signal int
 }
 
 type monitorChan chan GoSignal
 
-// A goroutine can be monitored.
+// A struct wrap goroutine to handle panic and can re-run easily.
 type Go struct {
 	lock sync.Mutex
 
@@ -46,7 +47,9 @@ func getNewRefId() int64 {
 }
 
 /*
-Create new Go.
+Create new Go struct.
+The first parameter is user function.
+The second parameter and more is parameter for parameter of user function.
 */
 func NewGo(fun any, params ...any) (ret *Go, retErr error) {
 	if retErr = verifyFunc(fun); retErr != nil {
@@ -65,6 +68,7 @@ func NewGo(fun any, params ...any) (ret *Go, retErr error) {
 
 /*
 Create Go and run.
+Same with function NewGo but run after create Go struct.
 */
 func NewGoAndRun(fun any, params ...any) (ret *Go, retErr error) {
 	ret, retErr = NewGo(fun, params...)

supervisor_test.go

@@ -18,7 +18,6 @@ func simpleLoopNoArg() {
 	for i := 0; i < 50; i++ {
 		time.Sleep(time.Millisecond)
 	}
-
 }
 
 func loopRun(a int, testSupporter chan int) {

worker.go

@@ -2,7 +2,6 @@ package easyworker
 
 import (
 	"log"
-	"reflect"
 	"time"
 )
 
@@ -65,7 +64,7 @@ func (w *worker) run() {
 
 	var (
 		task msg
-		ret  reflect.Value
+		ret  []any
 		err  error
 	)