test: add test cases for Paralleler.

Author: ghosind

paralleler.go

@@ -5,13 +5,16 @@ import (
 	"sync"
 )
 
+// Paralleler is a tool to run the tasks with the specific concurrency, default no concurrency
+// limitation.
 type Paralleler struct {
 	concurrency int
 	ctx         context.Context
 	locker      sync.Mutex
 	tasks       []AsyncFn
 }
 
+// WithConcurrency sets the number of concurrency limitation.
 func (p *Paralleler) WithConcurrency(concurrency int) *Paralleler {
 	if concurrency < 0 {
 		panic(ErrInvalidConcurrency)
@@ -22,12 +25,14 @@ func (p *Paralleler) WithConcurrency(concurrency int) *Paralleler {
 	return p
 }
 
+// WithContext sets the context that passes to the tasks.
 func (p *Paralleler) WithContext(ctx context.Context) *Paralleler {
 	p.ctx = ctx
 
 	return p
 }
 
+// Add adds the functions into the pending tasks list.
 func (p *Paralleler) Add(funcs ...AsyncFn) *Paralleler {
 	validateAsyncFuncs(funcs...)
 
@@ -39,6 +44,18 @@ func (p *Paralleler) Add(funcs ...AsyncFn) *Paralleler {
 	return p
 }
 
+// Clear clears the paralleler's pending tasks list.
+func (p *Paralleler) Clear() *Paralleler {
+	p.locker.Lock()
+	defer p.locker.Unlock()
+
+	p.tasks = nil
+
+	return p
+}
+
+// Run runs the tasks in the paralleler's pending list, it'll clear the pending list and return
+// the results of the tasks.
 func (p *Paralleler) Run() ([][]any, error) {
 	tasks := p.getTasks()
 	out := make([][]any, len(tasks))
@@ -74,6 +91,8 @@ func (p *Paralleler) Run() ([][]any, error) {
 	return out, nil
 }
 
+// getConcurrencyChan creates and returns a concurrency controlling channel by the specific number
+// of the concurrency limitation.
 func (p *Paralleler) getConcurrencyChan() chan empty {
 	var conch chan empty
 
@@ -84,6 +103,8 @@ func (p *Paralleler) getConcurrencyChan() chan empty {
 	return conch
 }
 
+// getTasks returns the tasks from the pending list, and clear the pending list to receiving new
+// tasks.
 func (p *Paralleler) getTasks() []AsyncFn {
 	p.locker.Lock()
 
@@ -95,6 +116,7 @@ func (p *Paralleler) getTasks() []AsyncFn {
 	return tasks
 }
 
+// runTasks runs the tasks with the concurrency limitation.
 func (p *Paralleler) runTasks(ctx context.Context, resCh chan executeResult, tasks []AsyncFn) {
 	conch := p.getConcurrencyChan()
 
@@ -107,6 +129,7 @@ func (p *Paralleler) runTasks(ctx context.Context, resCh chan executeResult, tas
 	}
 }
 
+// runTask runs the task function, and
 func (p *Paralleler) runTask(
 	ctx context.Context,
 	n int,

paralleler_test.go

@@ -0,0 +1,145 @@
+package async_test
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ghosind/go-assert"
+	"github.com/ghosind/go-async"
+)
+
+func TestParalleler(t *testing.T) {
+	a := assert.New(t)
+	cnt := atomic.Int32{}
+
+	p := new(async.Paralleler)
+	for i := 0; i < 5; i++ {
+		p.Add(func() {
+			cnt.Add(1)
+		})
+	}
+
+	a.EqualNow(cnt.Load(), 0)
+	_, err := p.Run()
+	a.Nil(err)
+	a.EqualNow(cnt.Load(), 5)
+}
+
+func TestParallelerAddTasks(t *testing.T) {
+	a := assert.New(t)
+	cnt := atomic.Int32{}
+
+	p := new(async.Paralleler)
+	for i := 0; i < 5; i++ {
+		p.Add(func() {
+			cnt.Add(1)
+		})
+	}
+
+	_, err := p.Run()
+	a.Nil(err)
+	a.EqualNow(cnt.Load(), 5)
+
+	for i := 0; i < 3; i++ {
+		p.Add(func() {
+			cnt.Add(1)
+		})
+	}
+
+	_, err = p.Run()
+	a.Nil(err)
+	a.EqualNow(cnt.Load(), 8)
+}
+
+func TestParallelerClear(t *testing.T) {
+	a := assert.New(t)
+	cnt := atomic.Int32{}
+
+	p := new(async.Paralleler)
+	for i := 0; i < 5; i++ {
+		p.Add(func() {
+			cnt.Add(1)
+		})
+	}
+
+	p.Clear()
+
+	for i := 0; i < 3; i++ {
+		p.Add(func() {
+			cnt.Add(1)
+		})
+	}
+
+	_, err := p.Run()
+	a.Nil(err)
+	a.EqualNow(cnt.Load(), 3)
+}
+
+func TestParallelerWithConcurrency(t *testing.T) {
+	a := assert.New(t)
+	cnt := atomic.Int32{}
+
+	p := new(async.Paralleler).WithConcurrency(2)
+	for i := 0; i < 5; i++ {
+		p.Add(func() {
+			time.Sleep(50 * time.Millisecond)
+			cnt.Add(1)
+		})
+	}
+
+	start := time.Now()
+	_, err := p.Run()
+	a.Nil(err)
+	a.EqualNow(cnt.Load(), 5)
+
+	dur := time.Since(start)
+	a.GtNow(dur, 150*time.Millisecond)
+	a.LtNow(dur, 200*time.Millisecond)
+}
+
+func TestParallelerWithContext(t *testing.T) {
+	a := assert.New(t)
+	cnt := atomic.Int32{}
+
+	ctx, canFunc := context.WithTimeout(context.Background(), 80*time.Millisecond)
+	defer canFunc()
+
+	p := new(async.Paralleler).WithConcurrency(2).WithContext(ctx)
+	for i := 0; i < 5; i++ {
+		p.Add(func(ctx context.Context) {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				time.Sleep(50 * time.Millisecond)
+				cnt.Add(1)
+			}
+		})
+	}
+
+	_, err := p.Run()
+	a.EqualNow(err, async.ErrContextCanceled)
+	a.EqualNow(cnt.Load(), 2)
+}
+
+func ExampleParalleler() {
+	p := new(async.Paralleler)
+
+	p.Add(func() int {
+		return 1
+	}).Add(func() int {
+		return 2
+	}).Add(func() string {
+		return "Hello"
+	})
+
+	ret, err := p.Run()
+	fmt.Println(ret)
+	fmt.Println(err)
+	// Output:
+	// [[1] [2] [Hello]]
+	// <nil>
+}