Added series and waterfall [goroutine for sequential tasks?? needs a solid answer]

Author: roshangade

async.go

@@ -6,9 +6,14 @@
 package async
 
 type Task func() interface{}
+type STask func(interface{}) interface{}
 type Channel chan interface{}
 type Result []interface{}
 
+// Reference: https://blog.golang.org/pipelines
+// Reference: https://en.wikipedia.org/wiki/Moore%27s_law
+// https://en.wikipedia.org/wiki/Amdahl%27s_law
+
 type Promise struct {
 	//TODO:
 	//Context - Cancel, Timeout
@@ -17,9 +22,12 @@ type Promise struct {
 	//Can I use map for tasks? It will be easy to associate results with keys
 }
 
-func (p *Promise) All(tasks []Task) Result {
-	// can not use buffer channels, because not able to maintain sequence
-	// using channel slice
+//
+// parallel execution of all tasks
+//
+func (p *Promise) Parallel(tasks []Task) Result {
+	// can not use buffer channels, because not able to maintain output sequence
+	// using slice channels
 	workers := make([]Channel, 0, len(tasks))
 	for _, task := range tasks {
 		workers = append(workers, func() Channel {
@@ -37,5 +45,64 @@ func (p *Promise) All(tasks []Task) Result {
 	for _, result := range workers {
 		out = append(out, <-result)
 	}
+
 	return out
 }
+
+//
+// sequential execution of all tasks
+//
+func (p *Promise) Series(tasks []Task) (out interface{}, err error) {
+	// unbuffered channel
+	worker := make(Channel)
+	defer close(worker)
+
+	for _, task := range tasks {
+		// need to check the benefits of executing tasks on another go routine
+		go func(task Task) {
+			worker <- task()
+		}(task)
+
+		out = <-worker
+
+		if err = isError(out); err != nil {
+			out = nil
+			break
+		}
+	}
+
+	return
+}
+
+//
+// sequential execution of all tasks, but output of all tasks will be input for next task
+//
+func (p *Promise) Waterfall(tasks []STask) (out interface{}, err error) {
+	// unbuffered channel
+	worker := make(Channel)
+	defer close(worker)
+
+	for _, task := range tasks {
+
+		go func(task STask) {
+			worker <- task(out)
+		}(task)
+
+		out = <-worker
+
+		if err = isError(out); err != nil {
+			out = nil
+			break
+		}
+	}
+
+	return
+}
+
+func isError(val interface{}) (err error) {
+	switch val.(type) {
+	case error:
+		err = val.(error)
+	}
+	return
+}

examples/main.go renamed to examples/parallel/main.go

@@ -27,7 +27,7 @@ func main() {
 	}
 
 	fmt.Println("From: ", time.Now())
-	result := promise.All(tasks)
+	result := promise.Parallel(tasks)
 	fmt.Println("End: ", time.Now())
 
 	fmt.Println("Result: ", result)

examples/series/main.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/go-rs/async"
+)
+
+func main() {
+	fmt.Println("start")
+	var promise async.Promise
+
+	tasks := []async.Task{
+		func() interface{} {
+			fmt.Println("Running Hello..........")
+			time.Sleep(2 * time.Second)
+			fmt.Println("Completed Hello..........")
+			//return errors.New("innocent")
+			return "Hello"
+		},
+		func() interface{} {
+			fmt.Println("Running World..........")
+			time.Sleep(1 * time.Second)
+			fmt.Println("Completed World..........")
+			return "World"
+		},
+	}
+
+	fmt.Println("From: ", time.Now())
+	result, err := promise.Series(tasks)
+	fmt.Println("End: ", time.Now())
+
+	fmt.Println("Result: ", result, err)
+
+	fmt.Println("exit")
+}