successfully able to run simple workflows

Signed-off-by: Casey Lee <cplee@nektos.com>

act/common/executor.go

@@ -1,6 +1,7 @@
 package common
 
 import (
+	"context"
 	"fmt"
 
 	log "github.com/sirupsen/logrus"
@@ -25,26 +26,76 @@ func Warningf(format string, args ...interface{}) Warning {
 }
 
 // Executor define contract for the steps of a workflow
-type Executor func() error
+type Executor func(ctx context.Context) error
 
 // Conditional define contract for the conditional predicate
-type Conditional func() bool
+type Conditional func(ctx context.Context) bool
+
+// NewInfoExecutor is an executor that logs messages
+func NewInfoExecutor(format string, args ...interface{}) Executor {
+	return func(ctx context.Context) error {
+		logger := Logger(ctx)
+		logger.Infof(format, args...)
+		return nil
+	}
+}
 
 // NewPipelineExecutor creates a new executor from a series of other executors
 func NewPipelineExecutor(executors ...Executor) Executor {
-	return func() error {
-		for _, executor := range executors {
-			if executor == nil {
-				continue
+	if executors == nil {
+		return func(ctx context.Context) error {
+			return nil
+		}
+	}
+	var rtn Executor
+	for _, executor := range executors {
+		if rtn == nil {
+			rtn = executor
+		} else {
+			rtn = rtn.Then(executor)
+		}
+	}
+	return rtn
+}
+
+// NewConditionalExecutor creates a new executor based on conditions
+func NewConditionalExecutor(conditional Conditional, trueExecutor Executor, falseExecutor Executor) Executor {
+	return func(ctx context.Context) error {
+		if conditional(ctx) {
+			if trueExecutor != nil {
+				return trueExecutor(ctx)
 			}
-			err := executor()
-			if err != nil {
-				switch err.(type) {
-				case Warning:
-					log.Warning(err.Error())
-					return nil
-				default:
-					log.Debugf("%+v", err)
+		} else {
+			if falseExecutor != nil {
+				return falseExecutor(ctx)
+			}
+		}
+		return nil
+	}
+}
+
+// NewErrorExecutor creates a new executor that always errors out
+func NewErrorExecutor(err error) Executor {
+	return func(ctx context.Context) error {
+		return err
+	}
+}
+
+// NewParallelExecutor creates a new executor from a parallel of other executors
+func NewParallelExecutor(executors ...Executor) Executor {
+	return func(ctx context.Context) error {
+		errChan := make(chan error)
+
+		for _, executor := range executors {
+			go executor.ChannelError(errChan)(ctx)
+		}
+
+		for i := 0; i < len(executors); i++ {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case err := <-errChan:
+				if err != nil {
 					return err
 				}
 			}
@@ -53,48 +104,76 @@ func NewPipelineExecutor(executors ...Executor) Executor {
 	}
 }
 
-// NewConditionalExecutor creates a new executor based on conditions
-func NewConditionalExecutor(conditional Conditional, trueExecutor Executor, falseExecutor Executor) Executor {
-	return func() error {
-		if conditional() {
-			if trueExecutor != nil {
-				return trueExecutor()
-			}
-		} else {
-			if falseExecutor != nil {
-				return falseExecutor()
+// ChannelError sends error to errChan rather than returning error
+func (e Executor) ChannelError(errChan chan error) Executor {
+	return func(ctx context.Context) error {
+		errChan <- e(ctx)
+		return nil
+	}
+}
+
+// Then runs another executor if this executor succeeds
+func (e Executor) Then(then Executor) Executor {
+	return func(ctx context.Context) error {
+		err := e(ctx)
+		if err != nil {
+			switch err.(type) {
+			case Warning:
+				log.Warning(err.Error())
+			default:
+				log.Debugf("%+v", err)
+				return err
 			}
 		}
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		return then(ctx)
+	}
+}
+
+// If only runs this executor if conditional is true
+func (e Executor) If(conditional Conditional) Executor {
+	return func(ctx context.Context) error {
+		if conditional(ctx) {
+			return e(ctx)
+		}
 		return nil
 	}
 }
 
-func executeWithChan(executor Executor, errChan chan error) {
-	errChan <- executor()
+// IfNot only runs this executor if conditional is true
+func (e Executor) IfNot(conditional Conditional) Executor {
+	return func(ctx context.Context) error {
+		if !conditional(ctx) {
+			return e(ctx)
+		}
+		return nil
+	}
 }
 
-// NewErrorExecutor creates a new executor that always errors out
-func NewErrorExecutor(err error) Executor {
-	return func() error {
+// IfBool only runs this executor if conditional is true
+func (e Executor) IfBool(conditional bool) Executor {
+	return e.If(func(ctx context.Context) bool {
+		return conditional
+	})
+}
+
+// Finally adds an executor to run after other executor
+func (e Executor) Finally(finally Executor) Executor {
+	return func(ctx context.Context) error {
+		err := e(ctx)
+		err2 := finally(ctx)
+		if err2 != nil {
+			return fmt.Errorf("Error occurred running finally: %v (original error: %v)", err2, err)
+		}
 		return err
 	}
 }
 
-// NewParallelExecutor creates a new executor from a parallel of other executors
-func NewParallelExecutor(executors ...Executor) Executor {
-	return func() error {
-		errChan := make(chan error)
-
-		for _, executor := range executors {
-			go executeWithChan(executor, errChan)
-		}
-
-		for i := 0; i < len(executors); i++ {
-			err := <-errChan
-			if err != nil {
-				return err
-			}
-		}
-		return nil
+// Not return an inverted conditional
+func (c Conditional) Not() Conditional {
+	return func(ctx context.Context) bool {
+		return !c(ctx)
 	}
 }