add supervisor & refactor code

Author: manhvu

README.md

@@ -4,43 +4,38 @@ A Golang package for supporting worker supervisor model.
 The package help developer easy to run tasks.
 It's scalable with minimum effort.
 
+easyworker inspired by Erlang OTP.
+
 # Design
 
 The package has two main part:
 
-* The supervisor: Manage worker, give a task to worker & collect result.
-* The worker: Receive task, run task then send result to supervisor.
+* The supervisor: Manage child, give a task to child & collect result.
+* The child: Receive task, run task then send result to supervisor.
 
 ## Supervisor
 
 Start worker and moniter worker.
 Send task to worker and get result.
 
-## Worker
+## Child
 
 Run task with user's function and handle error.
 If user's function panic worker will check retry config and re-run if needed.
 
 # Guide
 
-easywork support 2 type of worker:
+easywork support 2 type of worker and a type of supervisor:
 
-* Task. Add a list of task and run worker
-* Stream. Start worker then push data to worker from channel
+* Task, Add a list of task and run worker.
+* Stream, Start worker then push data to worker from channel.
+* Supervisor, Start a supervisor for custom worker.
 
 ## EasyTask
 
 This is simple way to run parallel task.
 User doesn't to manage goroutine, channel,...
 
-## EasyStream
-
-This is used for streaming type.
-In this case, tasks are continuously send to worker by user's channel.
-Results will receive from other channle of user.
-
-# Example
-
 EasyTask example:
 
 ```go
@@ -52,66 +47,73 @@ fnSum = func(a ...int) int {
 	return sum
 }
 
+// number of workers
 numWorkers := 3
+
+// retry times
 retryTimes := 0
-retrySleep := 0
 
-config, _ := NewConfig(fnSum, numWorkers, retryTimes, retrySleep)
+// sleep time before re-run
+retrySleep := 0
 
-task, err := NewTask(config)
+// new config for EasyTask
+config, _ := easyworker.NewConfig(fnSum, numWorkers, retryTimes, retrySleep)
 
-if err != nil {
-    t.Error("cannot create task, ", err)
-    return
-}
+// new EasyTask
+task, _ := easyworker.NewTask(config)
 
+// add tasks
 myTask.AddTask(1, 2, 3)
 myTask.AddTask(3, 4, 5, 6, 7)
 
+// start workers
 r, e := myTask.Run()
+
 if e != nil {
     t.Error("run task failed, ", e)
 } else {
     fmt.Println("task result:", r)
 }
 ```
 
+## EasyStream
+
+This is used for streaming type.
+In this case, tasks are continuously send to worker by user's channel.
+Results will receive from other channle of user.
+
 EasyStream example:
 
 ```go
+// fun will do task
 fnStr = func (a int, suffix string) string {
 	if a%3 == 0 {
 		panic("panic from user func")
 	}
 	return fmt.Sprintf("%d_%s", a, suffix)
 }
 
-inCh := make(chan []interface{}, 1)
+inCh := make(chan []interface{})
 outCh := make(chan interface{})
 
 // number of workers = number of cpu cores (logical cores)
-config, _ := NewConfig(fnSum, DefaultNumWorker(), 3, 1000)
+config, _ := easyworker.NewConfig(fnSum, DefaultNumWorker(), 3, 1000)
 
 // test with stream.
-myStream, err := NewStream(config, inCh, outCh)
-if err != nil {
-    t.Error("create EasyWorker failed, ", err)
-}
+myStream, _ := easyworker.NewStream(config, inCh, outCh)
 
-e := myStream.Run()
-if e != nil {
-    t.Error("run stream task failed, ", e)
-} else {
-    fmt.Println("stream is running")
-}
+// start stream.
+myStream.Run()
 
+// receive data from stream.
 go func() {
     for {
         r := <-outCh
         fmt.Println("stream result: ", r)
     }
 }()
 
+// send data to stream.
 go func() {
     for i := 0; i < 15; i++ {
         input := []interface{}{i, "3"}
@@ -121,8 +123,60 @@ go func() {
     }
 }()
 
-time.Sleep(2 * time.Second)
+// wait for get result
+time.Sleep(time.Second)
 
-// Stop all worker
+// stop all worker
 myStream.Stop()
-```
+```
+
+## Supervisor
+
+This is used for generic purpose worker(child).
+Every children has a owner restart strategy.
+
+Currently, child has three type of restart strategy:
+
+* ALWAYS_RESTART, supervisor always restart child if it panic/done.
+* NORMAL_RESTART, supervisor will restart if child was panic.
+* NO_RESTART, supervisor will don't restart for any reason.
+
+Child will be started after add to supervisor.
+
+supervisor example:
+
+```go
+// example function need to run in child.
+loopRun1 = func(a int) {
+	for i := 0; i < a; i++ {
+		time.Sleep(time.Second)
+		fmt.Println("loop at", i)
+	}
+	fmt.Println("Loop exit..")
+}
+
+// example function run in child. It will panic if counter > 3
+LoopRunWithPanic = func(a int) {
+	for i := 0; i < a; i++ {
+		time.Sleep(time.Second)
+		fmt.Println("loop at", i)
+		if i > 3 {
+			panic("test loop with panic")
+		}
+	}
+    // maybe you won't see this.
+	fmt.Println("Loop exit..")
+}
+
+// create a supervisor
+sup := easyworker.NewSupervisor()
+
+// add direct child to supervisor.
+sup.NewChild(easyworker.NORMAL_RESTART, LoopRun, 5)
+
+// create a child
+child, _ := easyworker.NewChild(ALWAYS_RESTART, LoopRunWithPanic, 5)
+
+// add exists child.
+sup.AddChild(child)
+```

child.go

@@ -0,0 +1,118 @@
+package easyworker
+
+import (
+	"fmt"
+	"reflect"
+)
+
+const (
+	// Always restart child
+	ALWAYS_RESTART = iota
+
+	// Just restart if child got an error
+	NORMAL_RESTART
+
+	// No restart child
+	NO_RESTART
+)
+
+const (
+	CHILD_PANIC = iota
+	CHILD_DONE
+	CHILD_RUNING
+	CHILD_RESTARTING
+	CHILD_STOPPED
+)
+
+var (
+	// use to store last id of child. id is auto_increment.
+	childLastId int
+)
+
+type Child struct {
+	id           int
+	restart_type int
+	cmdCh        chan cmd
+	status       int
+
+	fun    interface{}
+	params []interface{}
+}
+
+func NewChild(restart int, fun interface{}, params ...interface{}) (ret Child, retErr error) {
+	if restart < ALWAYS_RESTART || restart > NO_RESTART {
+		retErr = fmt.Errorf("in correct restart type, input: %d", restart)
+		return
+	}
+
+	if retErr = verifyFunc(fun); retErr != nil {
+		return
+	}
+
+	childLastId++
+
+	return Child{
+		id:           childLastId,
+		restart_type: restart,
+		fun:          fun,
+		params:       params,
+	}, nil
+}
+
+func (c *Child) run() {
+	defer func() {
+		msg := cmd{
+			id:      c.id,
+			typeCmd: CHILD_DONE,
+		}
+		if r := recover(); r != nil {
+			fmt.Println(c.id, ", worker was panic, ", r)
+			msg.typeCmd = CHILD_PANIC
+			msg.data = r
+		}
+
+		c.cmdCh <- msg
+	}()
+
+	var (
+		ret reflect.Value
+		err error
+	)
+
+	c.status = CHILD_RUNING
+l:
+	for {
+		// call user define function.
+		ret, err = invokeFun(c.fun, c.params...)
+
+		switch c.restart_type {
+		case ALWAYS_RESTART:
+			if err != nil {
+				fmt.Println(c.id, "failed, reason:", err)
+
+			}
+			fmt.Println(c.id, "child re-run")
+			continue l
+		case NORMAL_RESTART:
+			if err == nil {
+				fmt.Println(c.id, "done, child no re-run")
+				break l
+			} else {
+				fmt.Println(c.id, "failed, child re-run, reason:", err)
+				continue l
+			}
+		case NO_RESTART:
+			if err != nil {
+				fmt.Println(c.id, "failed, no re-run, reason:", err)
+				break l
+			}
+		}
+
+	}
+
+	if err != nil {
+		fmt.Println(c.id, ", call function failed, error: ", err)
+	} else {
+		fmt.Println(c.id, ", function return ", ret)
+	}
+}

child_test.go

@@ -0,0 +1,29 @@
+package easyworker
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func LoopRun2(a int) {
+	for i := 0; i < a; i++ {
+		time.Sleep(time.Second)
+		fmt.Println("loop at", i)
+	}
+	fmt.Println("Loop exit..")
+}
+
+func TestChildIncorrectedParams(t *testing.T) {
+	_, err := NewChild(ALWAYS_RESTART, "LoopRun", 5)
+
+	if err == nil {
+		t.Error("missed checking function from user")
+	}
+
+	_, err = NewChild(1000, LoopRun, 5)
+
+	if err == nil {
+		t.Error("missed checking restart strategy from user")
+	}
+}