Include task runner

Author: rodrigo-brito

async_test.go

@@ -1,13 +1,12 @@
 package async_test
 
 import (
+	"context"
 	"errors"
 	"sync"
 	"testing"
 	"time"
 
-	"context"
-
 	"github.com/StudioSol/async"
 	. "github.com/smartystreets/goconvey/convey"
 )

runner.go

@@ -0,0 +1,105 @@
+package async
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+type Runner struct {
+	sync.Mutex
+	tasks      []Task
+	errs       []error
+	limit      int
+	timeout    time.Duration
+	waitErrors bool
+}
+
+// NewRunner creates a new task manager to control async functions
+func NewRunner(tasks ...Task) *Runner {
+	return &Runner{
+		tasks: tasks,
+		limit: len(tasks),
+	}
+}
+
+// WaitErrors if, if active, will wait for the error response from all functions
+func (r *Runner) WaitErrors() *Runner {
+	r.waitErrors = true
+	return r
+}
+
+// WithLimit defines an limit for concurrent tasks execution
+func (r *Runner) WithLimit(limit int) *Runner {
+	r.limit = limit
+	return r
+}
+
+// AllErrors returns all errors reported by functions
+func (r *Runner) AllErrors() []error {
+	return r.errs
+}
+
+// AllErrors returns all errors reported by functions
+func (r *Runner) registerErr(err error) {
+	r.Lock()
+	defer r.Unlock()
+	if err != nil {
+		r.errs = append(r.errs, err)
+	}
+}
+
+func wrapperChannel(ctx context.Context, task Task) chan error {
+	cerr := make(chan error, 1)
+	go func() {
+		cerr <- task(ctx)
+		close(cerr)
+	}()
+	return cerr
+}
+
+// Run starts the task manager and return the first error or nil if succeed
+func (r *Runner) Run(parentCtx context.Context) error {
+	ctx, cancel := context.WithCancel(parentCtx)
+	cerr := make(chan error, len(r.tasks))
+	queue := make(chan struct{}, r.limit)
+	var wg sync.WaitGroup
+	wg.Add(len(r.tasks))
+	for _, task := range r.tasks {
+		queue <- struct{}{}
+		go func(fn func(context.Context) error) {
+			defer func() {
+				<-queue
+				wg.Done()
+				safePanic(cerr)
+			}()
+
+			select {
+			case <-parentCtx.Done():
+			case err := <-wrapperChannel(ctx, fn):
+				cerr <- err
+				r.registerErr(err)
+			}
+		}(task)
+	}
+
+	go func() {
+		wg.Wait()
+		cancel()
+		close(cerr)
+	}()
+
+	var firstErr error
+	for err := range cerr {
+		if err != nil && firstErr == nil {
+			firstErr = err
+			if r.waitErrors {
+				continue
+			}
+			cancel()
+			return firstErr
+		}
+	}
+
+	return firstErr
+}

runner_test.go

@@ -0,0 +1,76 @@
+package async
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRunner_AllErrors(t *testing.T) {
+	expectErr := errors.New("fail")
+	runner := NewRunner(func(context.Context) error {
+		return expectErr
+	}).WaitErrors()
+	err := runner.Run(context.Background())
+	require.Equal(t, expectErr, err)
+	require.Len(t, runner.AllErrors(), 1)
+	require.Equal(t, expectErr, runner.AllErrors()[0])
+}
+
+func TestRunner_WaitErrors(t *testing.T) {
+	expectErrOne := errors.New("fail")
+	expectErrTwo := errors.New("fail")
+	runner := NewRunner(func(context.Context) error {
+		return expectErrOne
+	}, func(context.Context) error {
+		return expectErrTwo
+	}).WaitErrors()
+	err := runner.Run(context.Background())
+	require.False(t, err != expectErrOne && err != expectErrTwo)
+	require.Len(t, runner.AllErrors(), 2)
+}
+
+func TestRunner_Run(t *testing.T) {
+	calledFist := false
+	calledSecond := false
+	runner := NewRunner(func(context.Context) error {
+		calledFist = true
+		return nil
+	}, func(context.Context) error {
+		calledSecond = true
+		return nil
+	})
+	err := runner.Run(context.Background())
+	require.Nil(t, err)
+	require.True(t, calledFist)
+	require.True(t, calledSecond)
+}
+
+func TestRunner_WithLimit(t *testing.T) {
+	order := 1
+	runner := NewRunner(func(context.Context) error {
+		require.Equal(t, 1, order)
+		order++
+		return nil
+	}, func(context.Context) error {
+		require.Equal(t, 2, order)
+		order++
+		return nil
+	}).WithLimit(1)
+	err := runner.Run(context.Background())
+	require.Nil(t, err)
+}
+
+func TestRunner_ContextCancelled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	runner := NewRunner(func(context.Context) error {
+		cancel()
+		time.Sleep(time.Minute)
+		return nil
+	})
+	err := runner.Run(ctx)
+	require.Nil(t, err)
+}