feat: add functio array foreach

internal/controller/trigger/validation/suscription_test.go

@@ -18,12 +18,10 @@ import (
 	"context"
 	"testing"
 
-	"google.golang.org/protobuf/types/known/structpb"
-
 	ctrlpb "github.com/linkall-labs/vanus/proto/pkg/controller"
 	metapb "github.com/linkall-labs/vanus/proto/pkg/meta"
-
 	. "github.com/smartystreets/goconvey/convey"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 func TestSubscriptionRequestValidator(t *testing.T) {

internal/primitive/transform/action/action.go

@@ -143,3 +143,18 @@ var (
 	ErrExist     = fmt.Errorf("action have exist")
 	ErrArgNumber = fmt.Errorf("action arg number invalid")
 )
+
+type NestAction interface {
+	Action
+	InitAction(actions []Action) error
+}
+
+type NestActionImpl struct {
+	CommonAction
+	Actions []Action
+}
+
+func (c *NestActionImpl) InitAction(actions []Action) error {
+	c.Actions = actions
+	return nil
+}

internal/primitive/transform/action/array/foreach.go

@@ -0,0 +1,64 @@
+// Copyright 2023 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package array
+
+import (
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/common"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/pkg/errors"
+)
+
+// ["array_foreach","array root", function].
+type arrayForeachAction struct {
+	action.NestActionImpl
+}
+
+func NewArrayForeachAction() action.Action {
+	a := &arrayForeachAction{}
+	a.CommonAction = action.CommonAction{
+		ActionName: "ARRAY_FOREACH",
+		FixedArgs:  []arg.TypeList{[]arg.Type{arg.EventData}},
+	}
+	return a
+}
+
+func (a *arrayForeachAction) Init(args []arg.Arg) error {
+	a.TargetArg = args[0]
+	a.Args = args
+	a.ArgTypes = []common.Type{common.Array}
+	return nil
+}
+
+func (a *arrayForeachAction) Execute(ceCtx *context.EventContext) error {
+	args, err := a.RunArgs(ceCtx)
+	if err != nil {
+		return err
+	}
+	arrayValue, _ := args[0].([]interface{})
+	for i := range arrayValue {
+		newCtx := &context.EventContext{
+			Data: arrayValue[i],
+		}
+		for i := range a.Actions {
+			err = a.Actions[i].Execute(newCtx)
+			if err != nil {
+				return errors.Wrapf(err, "action %dst execute error", i+1)
+			}
+		}
+	}
+	return a.TargetArg.SetValue(ceCtx, arrayValue)
+}

internal/primitive/transform/action/array/foreach_test.go

@@ -0,0 +1,68 @@
+// Copyright 2023 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package array_test
+
+import (
+	stdJson "encoding/json"
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/array"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestReplaceArrayAction(t *testing.T) {
+	funcName := array.NewArrayForeachAction().Name()
+	Convey("test replace array valid", t, func() {
+		jsonStr := `{
+			  "array": [
+				{
+				  "name": "name1",
+				  "number": 1
+				},
+				{
+				  "name": "name2",
+				  "number": "2"
+				},
+				{
+				  "name": "name3",
+				  "number": "3"
+				}
+			  ]
+		}`
+		Convey("replace valid", func() {
+			a, err := runtime.NewAction([]interface{}{funcName, "$.data.array", []interface{}{"add_prefix", "@.name", "prefix"}})
+			So(err, ShouldBeNil)
+
+			e := cetest.MinEvent()
+			var data map[string]interface{}
+			err = stdJson.Unmarshal([]byte(jsonStr), &data)
+			So(err, ShouldBeNil)
+			err = a.Execute(&context.EventContext{
+				Event: &e,
+				Data:  data,
+			})
+			So(err, ShouldBeNil)
+			value, exist := data["array"]
+			So(exist, ShouldBeTrue)
+			So(len(value.([]interface{})), ShouldEqual, 3)
+			So(value.([]interface{})[0].(map[string]interface{})["name"], ShouldEqual, "prefixname1")
+			So(value.([]interface{})[1].(map[string]interface{})["name"], ShouldEqual, "prefixname2")
+			So(value.([]interface{})[2].(map[string]interface{})["name"], ShouldEqual, "prefixname3")
+		})
+	})
+}

internal/primitive/transform/action/render/array.go → internal/primitive/transform/action/array/render.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package render
+package array
 
 import (
 	"fmt"

internal/primitive/transform/action/render/array_test.go → internal/primitive/transform/action/array/render_test.go

@@ -12,21 +12,21 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package render_test
+package array_test
 
 import (
 	stdJson "encoding/json"
 	"testing"
 
 	cetest "github.com/cloudevents/sdk-go/v2/test"
-	"github.com/linkall-labs/vanus/internal/primitive/transform/action/render"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/array"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
 	. "github.com/smartystreets/goconvey/convey"
 )
 
 func TestRenderArrayAction(t *testing.T) {
-	funcName := render.NewRenderArrayAction().Name()
+	funcName := array.NewRenderArrayAction().Name()
 	Convey("test render array invalid", t, func() {
 		jsonStr := `{
 			  "array": [

internal/primitive/transform/arg/arg.go

@@ -83,6 +83,9 @@ func NewArg(arg interface{}) (Arg, error) {
 		if argLen >= 2 && argName[:2] == EventArgPrefix {
 			return newEventAttribute(argName)
 		}
+		if argLen >= 2 && argName[:2] == EventDataSubArgPrefix {
+			return newEventData(EventDataArgPrefix + "." + argName[2:]), nil
+		}
 		if argLen >= 3 && argName[0] == '<' && argName[argLen-1] == '>' && argName[1] != '@' {
 			return newDefine(argName), nil
 		}
@@ -91,6 +94,7 @@ func NewArg(arg interface{}) (Arg, error) {
 }
 
 const (
-	EventArgPrefix     = "$."
-	EventDataArgPrefix = EventArgPrefix + "data"
+	EventArgPrefix        = "$."
+	EventDataArgPrefix    = EventArgPrefix + "data"
+	EventDataSubArgPrefix = "@."
 )

internal/primitive/transform/arg/event.go

@@ -115,8 +115,7 @@ func (arg eventData) Evaluate(ceCtx *context.EventContext) (interface{}, error)
 }
 
 func (arg eventData) SetValue(ceCtx *context.EventContext, value interface{}) error {
-	util.SetData(ceCtx.Data, arg.path, value)
-	return nil
+	return util.SetData(ceCtx.Data, arg.path, value)
 }
 
 func (arg eventData) DeleteValue(ceCtx *context.EventContext) error {

internal/primitive/transform/runtime/action.go

@@ -15,8 +15,7 @@
 package runtime
 
 import (
-	"fmt"
-	stdStrs "strings"
+	"strings"
 
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
@@ -29,9 +28,9 @@ var actionMap = map[string]newAction{}
 
 func AddAction(actionFn newAction) error {
 	a := actionFn()
-	name := stdStrs.ToUpper(a.Name())
+	name := strings.ToUpper(a.Name())
 	if _, exist := actionMap[name]; exist {
-		return fmt.Errorf("action %s has exist", name)
+		return errors.Errorf("action %s has exist", name)
 	}
 	actionMap[name] = actionFn
 	return nil
@@ -40,36 +39,64 @@ func AddAction(actionFn newAction) error {
 func NewAction(command []interface{}) (action.Action, error) {
 	funcName, ok := command[0].(string)
 	if !ok {
-		return nil, fmt.Errorf("command name must be string")
+		return nil, errors.Errorf("command name must be string")
 	}
-	actionFn, exist := actionMap[stdStrs.ToUpper(funcName)]
+	actionFn, exist := actionMap[strings.ToUpper(funcName)]
 	if !exist {
-		return nil, fmt.Errorf("command %s not exist", funcName)
+		return nil, errors.Errorf("command %s not exist", funcName)
 	}
 	a := actionFn()
 	argNum := len(command) - 1
 	if argNum < a.Arity() {
-		return nil, fmt.Errorf("command %s arg number is not enough, it need %d but only have %d",
+		return nil, errors.Errorf("command %s arg number is not enough, it need %d but only have %d",
 			funcName, a.Arity(), argNum)
 	}
-	if argNum > a.Arity() && !a.IsVariadic() {
-		return nil, fmt.Errorf("command %s arg number is too many, it need %d but have %d", funcName, a.Arity(), argNum)
+	nestAction, isNestAction := a.(action.NestAction)
+	if !isNestAction {
+		if argNum > a.Arity() && !a.IsVariadic() {
+			return nil, errors.Errorf("command %s arg number is too many, it need %d but have %d", funcName, a.Arity(), argNum)
+		}
+	} else {
+		argNum = a.Arity()
 	}
 	args := make([]arg.Arg, argNum)
-	for i := 1; i < len(command); i++ {
-		_arg, err := arg.NewArg(command[i])
+	for i := 0; i < len(args); i++ {
+		index := i + 1
+		_arg, err := arg.NewArg(command[index])
 		if err != nil {
-			return nil, errors.Wrapf(err, "command %s arg %d is invalid", funcName, i)
+			return nil, errors.Wrapf(err, "command %s arg %d is invalid", funcName, index)
 		}
-		argType := a.ArgType(i - 1)
+		argType := a.ArgType(i)
 		if !argType.Contains(_arg) {
-			return nil, fmt.Errorf("command %s arg %d not support type %s", funcName, i, _arg.Type())
+			return nil, errors.Errorf("command %s arg %d not support type %s", funcName, index, _arg.Type())
 		}
-		args[i-1] = _arg
+		args[i] = _arg
 	}
 	err := a.Init(args)
 	if err != nil {
 		return nil, errors.Wrapf(err, "command %s init error", funcName)
 	}
+	if isNestAction {
+		actions := make([]action.Action, len(command)-1-argNum)
+		if len(actions) == 0 {
+			return nil, errors.Errorf("command %s arg number is not enough, lost function arg", funcName)
+		}
+		for i := 0; i < len(actions); i++ {
+			index := i + 1 + argNum
+			if arr, ok := command[index].([]interface{}); ok {
+				_a, err := NewAction(arr)
+				if err != nil {
+					return nil, errors.Wrapf(err, "action %s arg %d new action failed", funcName, index)
+				}
+				actions[i] = _a
+			} else {
+				return nil, errors.Errorf("arg %d is invalid", index)
+			}
+		}
+		err = nestAction.InitAction(actions)
+		if err != nil {
+			return nil, errors.Wrapf(err, "command %s init action error", funcName)
+		}
+	}
 	return a, nil
 }

internal/primitive/transform/runtime/action_bench_test.go

@@ -17,10 +17,9 @@ package runtime
 import (
 	"testing"
 
-	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
-
 	ce "github.com/cloudevents/sdk-go/v2"
 	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
 )
 
 func actionBenchmark(command []interface{}) func(b *testing.B) {

internal/primitive/transform/runtime/init.go

@@ -15,11 +15,11 @@
 package runtime
 
 import (
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/array"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/common"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/condition"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/datetime"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/math"
-	"github.com/linkall-labs/vanus/internal/primitive/transform/action/render"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/source"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/strings"
 	"github.com/linkall-labs/vanus/internal/primitive/transform/action/structs"
@@ -53,8 +53,9 @@ func init() {
 		strings.NewReplaceBetweenPositionsAction,
 		// condition
 		condition.NewConditionIfAction,
-		// render
-		render.NewRenderArrayAction,
+		// array
+		array.NewRenderArrayAction,
+		array.NewArrayForeachAction,
 		// common
 		common.NewLengthAction,
 		// source