Skip to content

Commit

Permalink
feat: add transformer function condition if (#353)
Browse files Browse the repository at this point in the history
Signed-off-by: delu <[email protected]>

Signed-off-by: delu <[email protected]>
  • Loading branch information
xdlbdy authored Dec 20, 2022
1 parent 87085db commit 3fb8336
Show file tree
Hide file tree
Showing 11 changed files with 243 additions and 43 deletions.
12 changes: 8 additions & 4 deletions internal/primitive/transform/action/action.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,11 @@ import (
"fmt"
"strings"

"github.com/linkall-labs/vanus/internal/primitive/transform/function"

"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/linkall-labs/vanus/internal/primitive/transform/function"
"github.com/pkg/errors"
)

Expand All @@ -46,7 +48,7 @@ type commonAction struct {
fn function.Function

args []arg.Arg
argTypes []function.Type
argTypes []common.Type
targetArg arg.Arg
}

Expand Down Expand Up @@ -91,7 +93,7 @@ func (a *commonAction) setArgTypes() error {
if len(a.args) > a.fn.Arity() && !a.fn.IsVariadic() {
return ErrArgNumber
}
argTypes := make([]function.Type, len(a.args))
argTypes := make([]common.Type, len(a.args))
for i := 0; i < len(a.args); i++ {
argTypes[i] = *a.fn.ArgType(i)
}
Expand Down Expand Up @@ -124,7 +126,7 @@ func (a *commonAction) runArgs(ceCtx *context.EventContext) ([]interface{}, erro
if err != nil {
return nil, errors.Wrapf(err, "arg %s evaluate error", _arg.Original())
}
v, err := function.Cast(value, a.argTypes[i])
v, err := common.Cast(value, a.argTypes[i])
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -177,6 +179,8 @@ func init() {
newAddPrefixAction,
newAddSuffixAction,
newReplaceWithRegexAction,
// condition
newConditionIfAction,
} {
if err := AddAction(fn); err != nil {
panic(err)
Expand Down
85 changes: 85 additions & 0 deletions internal/primitive/transform/action/condition_if_action.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
// Copyright 2022 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 action

import (
"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"
)

type conditionIfAction struct {
commonAction
}

// ["condition_if","$.targetPath","$.path","op","compareValue","trueValue","falseValue"]
// op must be string and only support ==,>=,>,<=,< .
func newConditionIfAction() Action {
return &conditionIfAction{
commonAction{
name: "CONDITION_IF",
fixedArgs: []arg.TypeList{arg.EventList, arg.All, arg.All, arg.All, arg.All, arg.All},
},
}
}

func (a *conditionIfAction) Init(args []arg.Arg) error {
a.targetArg = args[0]
a.args = args[1:]
return nil
}

func (a *conditionIfAction) Execute(ceCtx *context.EventContext) error {
v, err := a.args[1].Evaluate(ceCtx)
if err != nil {
return errors.Wrapf(err, "arg %s evaluate error", a.args[1].Original())
}
op, ok := v.(string)
if !ok {
return errors.New("op type must be string")
}
if op == "==" {
a.argTypes = []common.Type{common.String, common.String, common.String, common.Any, common.Any}
} else {
switch op {
case ">=", ">", "<=", "<":
a.argTypes = []common.Type{common.Number, common.String, common.Number, common.Any, common.Any}
default:
return errors.Errorf("not support op [%s]", op)
}
}
args, err := a.runArgs(ceCtx)
if err != nil {
return err
}
var result bool
switch op {
case "==":
result = args[0] == args[2]
case ">=":
result = args[0].(float64) >= args[2].(float64)
case ">":
result = args[0].(float64) > args[2].(float64)
case "<=":
result = args[0].(float64) <= args[2].(float64)
case "<":
result = args[0].(float64) < args[2].(float64)
}
if result {
return a.targetArg.SetValue(ceCtx, args[3])
}
return a.targetArg.SetValue(ceCtx, args[4])
}
103 changes: 103 additions & 0 deletions internal/primitive/transform/action/condition_if_action_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
// Copyright 2022 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 action

import (
"testing"

"github.com/linkall-labs/vanus/internal/primitive/transform/context"
. "github.com/smartystreets/goconvey/convey"
)

func TestConditionIfAction(t *testing.T) {
Convey("test condition if ==", t, func() {
Convey("test string", func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", "==", "test", true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", "test")
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldBeNil)
So(e.Extensions()["test2"], ShouldEqual, true)
})
Convey("test number", func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", "==", 123, true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", 123)
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldBeNil)
So(e.Extensions()["test2"], ShouldEqual, true)
})
})
Convey("test condition >=", t, func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", ">=", int32(123), true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", "456")
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldBeNil)
So(e.Extensions()["test2"], ShouldEqual, true)
})
Convey("test condition >", t, func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", ">", int32(123), true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", 456)
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldBeNil)
So(e.Extensions()["test2"], ShouldEqual, true)
})
Convey("test condition <=", t, func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", "<=", "123", true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", 456)
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldBeNil)
So(e.Extensions()["test2"], ShouldEqual, false)
})
Convey("test condition <", t, func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", "<", "123", true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", 456)
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldBeNil)
So(e.Extensions()["test2"], ShouldEqual, false)
})
Convey("test condition unDefine op invalid", t, func() {
a, err := NewAction([]interface{}{newConditionIfAction().Name(), "$.test2", "$.test", "<==", "123", true, false})
So(err, ShouldBeNil)
e := newEvent()
e.SetExtension("test", 456)
err = a.Execute(&context.EventContext{
Event: e,
})
So(err, ShouldNotBeNil)
})
}
9 changes: 3 additions & 6 deletions internal/primitive/transform/action/regex_action.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,10 @@ import (
"regexp"
"sync"

"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/linkall-labs/vanus/internal/primitive/transform/function"

"github.com/pkg/errors"

"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
)

type replaceWithRegexAction struct {
Expand All @@ -47,7 +44,7 @@ func newReplaceWithRegexAction() Action {
func (a *replaceWithRegexAction) Init(args []arg.Arg) error {
a.targetArg = args[0]
a.args = args
a.argTypes = []function.Type{function.String, function.String, function.String}
a.argTypes = []common.Type{common.String, common.String, common.String}
return nil
}

Expand Down
13 changes: 6 additions & 7 deletions internal/primitive/transform/action/struct_action.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,9 @@ package action
import (
"fmt"

"github.com/linkall-labs/vanus/internal/primitive/transform/context"

"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
"github.com/linkall-labs/vanus/internal/primitive/transform/function"
"github.com/linkall-labs/vanus/internal/primitive/transform/common"
"github.com/linkall-labs/vanus/internal/primitive/transform/context"
)

// ["delete", "key"].
Expand Down Expand Up @@ -63,7 +62,7 @@ func newCreateActionAction() Action {
func (a *createAction) Init(args []arg.Arg) error {
a.targetArg = args[0]
a.args = args[1:]
a.argTypes = []function.Type{function.Any}
a.argTypes = []common.Type{common.Any}
return nil
}

Expand Down Expand Up @@ -96,7 +95,7 @@ func newReplaceAction() Action {
func (a *replaceAction) Init(args []arg.Arg) error {
a.targetArg = args[0]
a.args = args[1:]
a.argTypes = []function.Type{function.Any}
a.argTypes = []common.Type{common.Any}
return nil
}

Expand Down Expand Up @@ -129,7 +128,7 @@ func newMoveActionAction() Action {
func (a *moveAction) Init(args []arg.Arg) error {
a.targetArg = args[1]
a.args = args[:1]
a.argTypes = []function.Type{function.Any}
a.argTypes = []common.Type{common.Any}
return nil
}

Expand Down Expand Up @@ -166,7 +165,7 @@ func newRenameActionAction() Action {
func (a *renameAction) Init(args []arg.Arg) error {
a.targetArg = args[1]
a.args = args[:1]
a.argTypes = []function.Type{function.Any}
a.argTypes = []common.Type{common.Any}
return nil
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.

package function
package common

import (
"fmt"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.

package function
package common

type Type uint8

Expand Down
10 changes: 6 additions & 4 deletions internal/primitive/transform/function/format_functions.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,14 @@ import (
"time"

"github.com/linkall-labs/vanus/internal/primitive/transform/function/util"

"github.com/linkall-labs/vanus/internal/primitive/transform/common"
)

var DateFormatFunction = function{
name: "DATE_FORMAT",
fixedArgs: []Type{String, String},
variadicArgs: TypePtr(String),
fixedArgs: []common.Type{common.String, common.String},
variadicArgs: common.TypePtr(common.String),
fn: func(args []interface{}) (interface{}, error) {
t, err := time.ParseInLocation(time.RFC3339, args[0].(string), time.UTC)
if err != nil {
Expand All @@ -43,8 +45,8 @@ var DateFormatFunction = function{

var UnixTimeFormatFunction = function{
name: "UNIX_TIME_FORMAT",
fixedArgs: []Type{Number, String},
variadicArgs: TypePtr(String),
fixedArgs: []common.Type{common.Number, common.String},
variadicArgs: common.TypePtr(common.String),
fn: func(args []interface{}) (interface{}, error) {
t := time.Unix(int64(args[0].(float64)), 0)
loc := time.UTC
Expand Down
10 changes: 6 additions & 4 deletions internal/primitive/transform/function/function.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,15 @@

package function

import "github.com/linkall-labs/vanus/internal/primitive/transform/common"

type Function interface {
// Name func name
Name() string
// Arity arg number
Arity() int
// ArgType arg type
ArgType(index int) *Type
ArgType(index int) *common.Type
// IsVariadic is exist variadic
IsVariadic() bool
// Execute cal func result
Expand All @@ -29,8 +31,8 @@ type Function interface {

type function struct {
name string
fixedArgs []Type
variadicArgs *Type
fixedArgs []common.Type
variadicArgs *common.Type
fn func(args []interface{}) (interface{}, error)
}

Expand All @@ -42,7 +44,7 @@ func (f function) Arity() int {
return len(f.fixedArgs)
}

func (f function) ArgType(index int) *Type {
func (f function) ArgType(index int) *common.Type {
if index < len(f.fixedArgs) {
return &f.fixedArgs[index]
}
Expand Down
Loading

0 comments on commit 3fb8336

Please sign in to comment.