pipeline.go

package main

import (
	"fmt"
	"context"
)

func reverse(s string) string {
	r := []rune(s)
	for i, j := 0, len(r)-1; i < len(r)/2; i, j = i+1, j-1 {
		r[i], r[j] = r[j], r[i]
	}
	return string(r)
}

// Task is the interface that represents a unit of work
type Task interface {
	Run(context.Context, interface{}) (interface{}, error)
}

// FuncTask is a function Task
type FuncTask func(context.Context, interface{}) (interface{}, error)

// Run execute this FuncTask function
func (f FuncTask) Run(ctx context.Context, i interface{}) (interface{}, error) {
	return f(ctx, i)
}

type echoTask struct {
	Message string
}

func (e echoTask) Run(ctx context.Context, i interface{}) (interface{}, error) {
	return e.Message, nil
}

type reverseTask struct{}

func (r reverseTask) Run(ctx context.Context, i interface{}) (interface{}, error) {
	if s, ok := i.(string); ok {
		return reverse(s), nil
	}

	return "", nil
}

type printTask struct {}

func (p printTask) Run(ctx context.Context, i interface{}) (interface{}, error) {
	fmt.Println(i)
	return nil, nil
}

type chain struct {
	tasks []Task
}

// Do execute all the jobs in the work Chain
func (c chain) Run(ctx context.Context, i interface{}) (interface{}, error) {
	var result interface{}
	
	for _, task := range c.tasks {
		var err error
		
		result, err = task.Run(ctx, result)
		
		if err != nil {
			return nil, err
		}
	}

	return nil, nil
}

// Chain create a new Job chain
func Chain(tasks ...Task) Task {
	return chain{tasks: tasks}
}

// Run runs a task chain
func Run(ctx context.Context, task Task) (interface{}, error) {
	return task.Run(ctx, nil)
}

func main() {
	pipeline := Chain(
		echoTask{"Hello World"},
		reverseTask{},
		printTask{},
		FuncTask(func(ctx context.Context, i interface{}) (interface{}, error) {
			fmt.Println("Finishing")
			return nil, nil
		}),
	)
	
	Run(context.Background(), pipeline)
}