add Periodic, NoError, Error and fix bugs

errs.go

@@ -1,6 +1,7 @@
 package conc
 
 import (
+	"context"
 	"strconv"
 	"strings"
 )
@@ -37,3 +38,18 @@ func (errs Errs) Sanitize() error {
 	}
 	return nil
 }
+
+// NoError wrap a Task and never returns error
+func NoError(task Task) Task {
+	return TaskFunc(func(ctx context.Context) error {
+		_ = task.Do(ctx)
+		return nil
+	})
+}
+
+// Error create Task always returns error
+func Error(err error) Task {
+	return TaskFunc(func(ctx context.Context) error {
+		return err
+	})
+}

errs_test.go

@@ -1,8 +1,10 @@
 package conc
 
 import (
+	"context"
 	"errors"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"testing"
 )
 
@@ -21,3 +23,12 @@ func TestErrs_Sanitize(t *testing.T) {
 	errs = nil
 	assert.Equal(t, nil, errs.Sanitize())
 }
+
+func TestNoError(t *testing.T) {
+	et := Error(errors.New("failed"))
+	err := et.Do(context.Background())
+	require.Error(t, err)
+	net := NoError(et)
+	err = net.Do(context.Background())
+	require.NoError(t, err)
+}

parallel.go

@@ -25,6 +25,7 @@ func Parallel(tasks ...Task) Task {
 			}()
 		}
 		wg.Wait()
+		ctx2Cancel()
 		return errs.Sanitize()
 	})
 }
@@ -55,6 +56,7 @@ func ParallelWithLimit(limit int, tasks ...Task) Task {
 			}()
 		}
 		wg.Wait()
+		ctx2Cancel()
 		return errs.Sanitize()
 	})
 }

periodic.go

@@ -0,0 +1,27 @@
+package conc
+
+import (
+	"context"
+	"time"
+)
+
+// Periodic returns a Task do input Task periodically
+func Periodic(task Task, dur time.Duration, init bool) Task {
+	return TaskFunc(func(ctx context.Context) (err error) {
+		if init {
+			if err = task.Do(ctx); err != nil {
+				return
+			}
+		}
+		for {
+			select {
+			case <-time.After(dur):
+			case <-ctx.Done():
+				return
+			}
+			if err = task.Do(ctx); err != nil {
+				return
+			}
+		}
+	})
+}

periodic_test.go

@@ -0,0 +1,65 @@
+package conc
+
+import (
+	"context"
+	"errors"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestPeriodic(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	var v int
+	task := TaskFunc(func(ctx context.Context) error {
+		v++
+		return nil
+	})
+	pt := Periodic(task, time.Millisecond*200, true)
+	go pt.Do(ctx)
+
+	time.Sleep(time.Millisecond * 500)
+	cancel()
+
+	assert.Equal(t, 3, v)
+}
+
+func TestPeriodicWithError(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	var v int
+	task := TaskFunc(func(ctx context.Context) error {
+		v++
+		return errors.New("failed")
+	})
+	var err error
+	pt := Periodic(task, time.Millisecond*200, true)
+	go func() { err = pt.Do(ctx) }()
+
+	time.Sleep(time.Millisecond * 500)
+	cancel()
+
+	assert.Equal(t, 1, v)
+	assert.Error(t, err)
+}
+
+func TestPeriodicWithError2(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	var v int
+	task := TaskFunc(func(ctx context.Context) error {
+		v++
+		if v == 2 {
+			return errors.New("failed")
+		} else {
+			return nil
+		}
+	})
+	var err error
+	pt := Periodic(task, time.Millisecond*200, true)
+	go func() { err = pt.Do(ctx) }()
+
+	time.Sleep(time.Millisecond * 500)
+	cancel()
+
+	assert.Equal(t, 2, v)
+	assert.Error(t, err)
+}