feat: add fx.ParallelErr (#4107)

core/fx/parallel.go

@@ -1,6 +1,9 @@
 package fx
 
-import "github.com/zeromicro/go-zero/core/threading"
+import (
+	"github.com/zeromicro/go-zero/core/errorx"
+	"github.com/zeromicro/go-zero/core/threading"
+)
 
 // Parallel runs fns parallelly and waits for done.
 func Parallel(fns ...func()) {
@@ -10,3 +13,20 @@ func Parallel(fns ...func()) {
 	}
 	group.Wait()
 }
+
+func ParallelErr(fns ...func() error) error {
+	var be errorx.BatchError
+
+	group := threading.NewRoutineGroup()
+	for _, fn := range fns {
+		f := fn
+		group.RunSafe(func() {
+			if err := f(); err != nil {
+				be.Add(err)
+			}
+		})
+	}
+	group.Wait()
+
+	return be.Err()
+}

core/fx/parallel_test.go

@@ -1,6 +1,7 @@
 package fx
 
 import (
+	"errors"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -22,3 +23,54 @@ func TestParallel(t *testing.T) {
 	})
 	assert.Equal(t, int32(6), count)
 }
+
+func TestParallelErr(t *testing.T) {
+	var count int32
+	err := ParallelErr(
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 1)
+			return errors.New("failed to exec #1")
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 2)
+			return errors.New("failed to exec #2")
+
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 3)
+			return nil
+		},
+	)
+
+	assert.Equal(t, int32(6), count)
+	assert.Error(t, err)
+	assert.ErrorContains(t, err, "failed to exec #1", "failed to exec #2")
+}
+
+func TestParallelErrErrorNil(t *testing.T) {
+	var count int32
+	err := ParallelErr(
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 1)
+			return nil
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 2)
+			return nil
+
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 3)
+			return nil
+		},
+	)
+
+	assert.Equal(t, int32(6), count)
+	assert.NoError(t, err)
+}