✨ Address review comments

Author: acabarbaye

changes/20250820140853.feature

@@ -0,0 +1 @@
+:sparkles: `[parallelisation]` Added new compound execution group to support nested execution groups

utils/parallelisation/contextual.go

@@ -11,6 +11,8 @@ func DetermineContextError(ctx context.Context) error {
 	return commonerrors.ConvertContextError(ctx.Err())
 }
 
+type ContextualFunc func(ctx context.Context) error
+
 type ContextualFunctionGroup struct {
 	ExecutionGroup[ContextualFunc]
 }

utils/parallelisation/contextual_test.go

@@ -20,10 +20,10 @@ func TestForEach(t *testing.T) {
 
 	t.Run("close with 1 error but error collection", func(t *testing.T) {
 		closeError := commonerrors.ErrUnexpected
-		errortest.AssertError(t, ForEach(context.Background(), WithOptions(Parallel, JoinErrors), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc), WrapCloseToContextualFunc(func() error { return closeError }), WrapCancelToContextualFunc(cancelFunc)), closeError)
+		errortest.AssertError(t, ForEach(context.Background(), WithOptions(Parallel), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc), WrapCloseToContextualFunc(func() error { return closeError }), WrapCancelToContextualFunc(cancelFunc)), closeError)
 	})
 
-	t.Run("close with 1 error but error collection", func(t *testing.T) {
+	t.Run("close with 1 error and limited number of parallel workers", func(t *testing.T) {
 		closeError := commonerrors.ErrUnexpected
 		errortest.AssertError(t, ForEach(context.Background(), WithOptions(Workers(5), JoinErrors), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc), WrapCloseToContextualFunc(func() error { return closeError }), WrapCancelToContextualFunc(cancelFunc)), closeError)
 	})
@@ -45,4 +45,7 @@ func TestForEach(t *testing.T) {
 	t.Run("break on error with no error", func(t *testing.T) {
 		require.NoError(t, BreakOnError(context.Background(), WithOptions(Workers(5), JoinErrors), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc)))
 	})
+	t.Run("for each with no error", func(t *testing.T) {
+		require.NoError(t, ForEach(context.Background(), WithOptions(Workers(5), JoinErrors), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc), WrapCancelToContextualFunc(cancelFunc)))
+	})
 }

utils/parallelisation/group.go

@@ -23,27 +23,49 @@ type StoreOptions struct {
 	workers          int
 }
 
+func (o *StoreOptions) Default() *StoreOptions {
+	o.clearOnExecution = false
+	o.stopOnFirstError = false
+	o.sequential = false
+	o.reverse = false
+	o.joinErrors = false
+	o.onlyOnce = false
+	o.workers = 0
+	return o
+}
+
 func (o *StoreOptions) Merge(opts *StoreOptions) *StoreOptions {
 	if opts == nil {
 		return o
 	}
-	return &StoreOptions{
-		clearOnExecution: opts.clearOnExecution || o.clearOnExecution,
-		stopOnFirstError: opts.stopOnFirstError || o.stopOnFirstError,
-		sequential:       opts.sequential || o.sequential,
-		reverse:          opts.reverse || o.reverse,
-		joinErrors:       opts.joinErrors || o.joinErrors,
-		onlyOnce:         opts.onlyOnce || o.onlyOnce,
-		workers:          safecast.ToInt(math.Max(float64(opts.workers), float64(o.workers))),
-	}
+	o.clearOnExecution = opts.clearOnExecution || o.clearOnExecution
+	o.stopOnFirstError = opts.stopOnFirstError || o.stopOnFirstError
+	o.sequential = opts.sequential || o.sequential
+	o.reverse = opts.reverse || o.reverse
+	o.joinErrors = opts.joinErrors || o.joinErrors
+	o.onlyOnce = opts.onlyOnce || o.onlyOnce
+	o.workers = safecast.ToInt(math.Max(float64(opts.workers), float64(o.workers)))
+	return o
+}
+
+func (o *StoreOptions) MergeWithOptions(opt ...StoreOption) *StoreOptions {
+	return o.Merge(WithOptions(opt...))
+}
+
+func (o *StoreOptions) Overwrite(opts *StoreOptions) *StoreOptions {
+	return o.Default().Merge(opts)
+}
+
+func (o *StoreOptions) WithOptions(opts ...StoreOption) *StoreOptions {
+	return o.Overwrite(WithOptions(opts...))
 }
 
 func (o *StoreOptions) Options() []StoreOption {
 	return []StoreOption{
 		func(opts *StoreOptions) *StoreOptions {
 			op := o
 			if op == nil {
-				op = &StoreOptions{}
+				op = DefaultOptions()
 			}
 			return op.Merge(opts)
 		},
@@ -55,7 +77,7 @@ type StoreOption func(*StoreOptions) *StoreOptions
 // StopOnFirstError stops ExecutionGroup execution on first error.
 var StopOnFirstError StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.stopOnFirstError = true
 	o.joinErrors = false
@@ -66,7 +88,7 @@ var StopOnFirstError StoreOption = func(o *StoreOptions) *StoreOptions {
 // This option should not be used in combination to StopOnFirstError.
 var JoinErrors StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.stopOnFirstError = false
 	o.joinErrors = true
@@ -76,7 +98,7 @@ var JoinErrors StoreOption = func(o *StoreOptions) *StoreOptions {
 // OnlyOnce will ensure the function are executed only once if they do.
 var OnlyOnce StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.onlyOnce = true
 	return o
@@ -85,7 +107,7 @@ var OnlyOnce StoreOption = func(o *StoreOptions) *StoreOptions {
 // AnyTimes will allow the functions to be executed as often that they might be.
 var AnyTimes StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.onlyOnce = false
 	return o
@@ -94,7 +116,7 @@ var AnyTimes StoreOption = func(o *StoreOptions) *StoreOptions {
 // ExecuteAll executes all functions in the ExecutionGroup even if an error is raised. the first error raised is then returned.
 var ExecuteAll StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.stopOnFirstError = false
 	return o
@@ -103,7 +125,7 @@ var ExecuteAll StoreOption = func(o *StoreOptions) *StoreOptions {
 // ClearAfterExecution clears the ExecutionGroup after execution.
 var ClearAfterExecution StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.clearOnExecution = true
 	return o
@@ -112,7 +134,7 @@ var ClearAfterExecution StoreOption = func(o *StoreOptions) *StoreOptions {
 // RetainAfterExecution keep the ExecutionGroup intact after execution (no reset).
 var RetainAfterExecution StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.clearOnExecution = false
 	return o
@@ -121,7 +143,7 @@ var RetainAfterExecution StoreOption = func(o *StoreOptions) *StoreOptions {
 // Parallel ensures every function registered in the ExecutionGroup is executed concurrently in the order they were registered.
 var Parallel StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.sequential = false
 	return o
@@ -131,7 +153,7 @@ var Parallel StoreOption = func(o *StoreOptions) *StoreOptions {
 func Workers(workers int) StoreOption {
 	return func(o *StoreOptions) *StoreOptions {
 		if o == nil {
-			o = &StoreOptions{}
+			o = DefaultOptions()
 		}
 		o.workers = workers
 		o.sequential = false
@@ -142,7 +164,7 @@ func Workers(workers int) StoreOption {
 // Sequential ensures every function registered in the ExecutionGroup is executed sequentially in the order they were registered.
 var Sequential StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.sequential = true
 	return o
@@ -151,7 +173,7 @@ var Sequential StoreOption = func(o *StoreOptions) *StoreOptions {
 // SequentialInReverse ensures every function registered in the ExecutionGroup is executed sequentially but in the reverse order they were registered.
 var SequentialInReverse StoreOption = func(o *StoreOptions) *StoreOptions {
 	if o == nil {
-		o = &StoreOptions{}
+		o = DefaultOptions()
 	}
 	o.sequential = true
 	o.reverse = true
@@ -164,11 +186,34 @@ func WithOptions(option ...StoreOption) (opts *StoreOptions) {
 		opts = option[i](opts)
 	}
 	if opts == nil {
-		opts = &StoreOptions{}
+		opts = DefaultOptions()
 	}
 	return
 }
 
+// DefaultOptions returns the default store configuration
+func DefaultOptions() *StoreOptions {
+	opts := &StoreOptions{}
+	return opts.Default()
+}
+
+type IExecutor interface {
+	// Execute executes all the functions in the group.
+	Execute(ctx context.Context) error
+}
+
+type IExecutionGroup[T any] interface {
+	IExecutor
+	RegisterFunction(function ...T)
+	Len() int
+}
+
+type ICompoundExecutionGroup[T any] interface {
+	IExecutionGroup[T]
+	// RegisterExecutor registers executors of any kind to the group: they could be functions or sub-groups.
+	RegisterExecutor(executor ...IExecutor)
+}
+
 // NewExecutionGroup returns an execution group which executes functions according to store options.
 func NewExecutionGroup[T any](executeFunc ExecuteFunc[T], options ...StoreOption) *ExecutionGroup[T] {
 
@@ -366,3 +411,25 @@ func newWrapped[T any](e T, once bool) wrappedElement[T] {
 		return newBasicWrap[T](e)
 	}
 }
+
+var _ ICompoundExecutionGroup[ContextualFunc] = &CompoundExecutionGroup{}
+
+// NewCompoundExecutionGroup returns an execution group made of executors
+func NewCompoundExecutionGroup(options ...StoreOption) *CompoundExecutionGroup {
+	return &CompoundExecutionGroup{
+		ContextualFunctionGroup: *NewContextualGroup(options...),
+	}
+}
+
+type CompoundExecutionGroup struct {
+	ContextualFunctionGroup
+}
+
+// RegisterExecutor registers executors
+func (g *CompoundExecutionGroup) RegisterExecutor(group ...IExecutor) {
+	for i := range group {
+		g.RegisterFunction(func(ctx context.Context) error {
+			return group[i].Execute(ctx)
+		})
+	}
+}

utils/parallelisation/group_test.go

@@ -1,11 +1,15 @@
 package parallelisation
 
 import (
+	"context"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 
+	"github.com/ARM-software/golang-utils/utils/commonerrors"
+	"github.com/ARM-software/golang-utils/utils/commonerrors/errortest"
 	"github.com/ARM-software/golang-utils/utils/parallelisation/mocks"
 )
 
@@ -146,3 +150,43 @@ func TestExecutionTimes(t *testing.T) {
 		require.NoError(t, group.Close())
 	})
 }
+
+func TestCompoundGroup(t *testing.T) {
+	ctlr := gomock.NewController(t)
+	defer ctlr.Finish()
+
+	closerMock := mocks.NewMockCloser(ctlr)
+	closerMock.EXPECT().Close().Return(nil).Times(17)
+	group := NewCloserStoreWithOptions(ExecuteAll, OnlyOnce, Sequential)
+	group.RegisterFunction(closerMock, closerMock, closerMock)
+
+	compoundGroup := NewCompoundExecutionGroup(Parallel, RetainAfterExecution)
+	compoundGroup.RegisterFunction(WrapCloseToContextualFunc(WrapCloserIntoCloseFunc(closerMock)))
+	compoundGroup.RegisterExecutor(group)
+	compoundGroup.RegisterFunction(WrapCancelToContextualFunc(WrapContextualToCancelFunc(WrapCloseToContextualFunc(WrapCloserIntoCloseFunc(closerMock)))))
+
+	assert.Equal(t, 3, compoundGroup.Len())
+
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+	require.NoError(t, compoundGroup.Execute(context.Background()))
+
+	t.Run("With cancelled context", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		errortest.AssertError(t, compoundGroup.Execute(ctx), commonerrors.ErrCancelled)
+	})
+
+}
+
+func TestStoreOptions_MergeWithOptions(t *testing.T) {
+	opts := WithOptions(Parallel).MergeWithOptions(OnlyOnce, ExecuteAll, Workers(5), Sequential)
+	assert.True(t, opts.onlyOnce)
+	assert.False(t, opts.stopOnFirstError)
+	assert.True(t, opts.sequential)
+	assert.Equal(t, 5, opts.workers)
+}

utils/parallelisation/onclose.go

@@ -29,17 +29,14 @@ func NewCloserStore(stopOnFirstError bool) *CloserStore {
 	if stopOnFirstError {
 		option = StopOnFirstError
 	}
-	return NewCloserStoreWithOptions(option, Parallel, RetainAfterExecution)
+	return NewCloserStoreWithOptions(option, Parallel, OnlyOnce, RetainAfterExecution)
 }
 
 // NewCloserStoreWithOptions returns a store of io.Closer object which will all be closed on Close(). The first error received if any will be returned
 func NewCloserStoreWithOptions(opts ...StoreOption) *CloserStore {
 	return &CloserStore{
-		ExecutionGroup: *NewExecutionGroup[io.Closer](func(_ context.Context, closerObj io.Closer) error {
-			if closerObj == nil {
-				return commonerrors.UndefinedVariable("closer object")
-			}
-			return closerObj.Close()
+		ExecutionGroup: *NewExecutionGroup[io.Closer](func(ctx context.Context, closerObj io.Closer) error {
+			return WrapCloseToContextualFunc(WrapCloserIntoCloseFunc(closerObj))(ctx)
 		}, opts...),
 	}
 }
@@ -86,9 +83,21 @@ func CloseAllFuncAndCollateErrors(cs ...CloseFunc) error {
 	return group.Close()
 }
 
-type ContextualFunc func(ctx context.Context) error
 type CloseFunc func() error
 
+func (c CloseFunc) Close() error {
+	return c()
+}
+
+func WrapCloserIntoCloseFunc(closer io.Closer) CloseFunc {
+	return func() error {
+		if closer == nil {
+			return commonerrors.UndefinedVariable("closer object")
+		}
+		return closer.Close()
+	}
+}
+
 func WrapCancelToCloseFunc(f context.CancelFunc) CloseFunc {
 	return func() error {
 		f()
@@ -172,5 +181,10 @@ func NewConcurrentCloseFunctionStore(stopOnFirstError bool) *CloseFunctionStore
 	if stopOnFirstError {
 		option = StopOnFirstError
 	}
-	return NewCloseFunctionStore(option, Parallel, RetainAfterExecution)
+	return NewCloseFunctionStore(option, Parallel, RetainAfterExecution, OnlyOnce)
+}
+
+// NewCloseOnceGroup is the same as NewCloseFunctionStore but ensures any closing functions are only executed once.
+func NewCloseOnceGroup(options ...StoreOption) *CloseFunctionStore {
+	return NewCloseFunctionStore(OnlyOnce(WithOptions(options...)).Options()...)
 }

utils/parallelisation/onclose_test.go

@@ -72,6 +72,21 @@ func TestCloseAll(t *testing.T) {
 
 }
 
+func TestCloseOnce(t *testing.T) {
+	t.Run("close every function once", func(t *testing.T) {
+		ctlr := gomock.NewController(t)
+		defer ctlr.Finish()
+		closeError := commonerrors.ErrUnexpected
+
+		closerMock := mocks.NewMockCloser(ctlr)
+		closerMock.EXPECT().Close().Return(closeError).Times(3)
+
+		group := NewCloseOnceGroup(Parallel, RetainAfterExecution)
+		group.RegisterCloseFunction(WrapCloserIntoCloseFunc(closerMock), WrapCloserIntoCloseFunc(closerMock), WrapCloserIntoCloseFunc(closerMock))
+		errortest.AssertError(t, group.Close(), closeError)
+	})
+}
+
 func TestCancelOnClose(t *testing.T) {
 	t.Run("parallel", func(t *testing.T) {
 		closeStore := NewCloseFunctionStoreStore(true)
@@ -136,7 +151,7 @@ func TestSequentialExecution(t *testing.T) {
 	for i := range tests {
 		test := tests[i]
 		t.Run(fmt.Sprintf("%v-%#v", i, test.option), func(t *testing.T) {
-			opt := test.option(&StoreOptions{})
+			opt := test.option(DefaultOptions())
 			t.Run("sequentially", func(t *testing.T) {
 				closeStore := NewCloseFunctionStore(test.option, Sequential)
 				ctx1, cancel1 := context.WithCancel(context.Background())