diff --git a/apier/v1/apier.go b/apier/v1/apier.go index 56d5ce27f..94434b5bb 100644 --- a/apier/v1/apier.go +++ b/apier/v1/apier.go @@ -104,11 +104,12 @@ func (self *ApierV1) GetRatingPlan(rplnId string, reply *engine.RatingPlan) erro } func (self *ApierV1) ExecuteAction(attr *utils.AttrExecuteAction, reply *string) error { - accID := utils.AccountKey(attr.Tenant, attr.Account) at := &engine.ActionTiming{ ActionsID: attr.ActionsId, } - at.SetAccountIDs(utils.StringMap{accID: true}) + if attr.Tenant != "" && attr.Account != "" { + at.SetAccountIDs(utils.StringMap{utils.AccountKey(attr.Tenant, attr.Account): true}) + } if err := at.Execute(); err != nil { *reply = err.Error() return err diff --git a/apier/v1/apier_local_test.go b/apier/v1/apier_local_test.go index 5c2504cec..124b6a984 100644 --- a/apier/v1/apier_local_test.go +++ b/apier/v1/apier_local_test.go @@ -1297,7 +1297,7 @@ func TestApierResetDataAfterLoadFromFolder(t *testing.T) { if rcvStats.Destinations != 5 || rcvStats.RatingPlans != 5 || rcvStats.RatingProfiles != 5 || - rcvStats.Actions != 9 || + rcvStats.Actions != 10 || rcvStats.DerivedChargers != 3 { t.Errorf("Calling ApierV1.GetCacheStats received: %+v", rcvStats) } diff --git a/data/tariffplans/testtp/Actions.csv b/data/tariffplans/testtp/Actions.csv index 98a48be14..1525522f2 100644 --- a/data/tariffplans/testtp/Actions.csv +++ b/data/tariffplans/testtp/Actions.csv @@ -9,3 +9,4 @@ TOPUP_DATA_r,*topup,,,,*monetary,*out,,DATA_DEST,,,*unlimited,,5000000,10,false, TOPUP_DATA_r,*topup,,,,*data,*out,,DATA_DEST,datar,,*unlimited,,50000000000,10,false,false,10 TOPUP_VOICE,*topup,,,,*voice,*out,,GERMANY_MOBILE,,,*unlimited,,50000,10,false,false,10 TOPUP_NEG,*topup,,,,*voice,*out,,GERMANY;!GERMANY_MOBILE,*zero1m,,*unlimited,,100,10,false,false,10 +RPC,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""ApierV2.SetAccount"",""Attempts"":1,""Async"" :false,""Param"":{""Account"":""rpc"",""Tenant"":""cgrates.org""}}",,,,,,,,,,,,,,, diff --git a/engine/action.go b/engine/action.go index 62ca439dd..84669ff1f 100644 --- a/engine/action.go +++ b/engine/action.go @@ -32,6 +32,7 @@ import ( "github.com/cgrates/cgrates/config" "github.com/cgrates/cgrates/utils" + "github.com/cgrates/rpcclient" ) /* @@ -74,6 +75,7 @@ const ( CDRLOG = "*cdrlog" SET_DDESTINATIONS = "*set_ddestinations" TRANSFER_MONETARY_DEFAULT = "*transfer_monetary_default" + CGR_RPC = "*cgr_rpc" ) func (a *Action) Clone() *Action { @@ -140,6 +142,8 @@ func getActionFunc(typ string) (actionTypeFunc, bool) { return setBalanceAction, true case TRANSFER_MONETARY_DEFAULT: return transferMonetaryDefaultAction, true + case CGR_RPC: + return cgrRPCAction, true } return nil, false } @@ -647,6 +651,53 @@ func transferMonetaryDefaultAction(acc *Account, sq *StatsQueueTriggered, a *Act return nil } +type RPCRequest struct { + Address string + Transport string + Method string + Attempts int + Async bool + Param map[string]interface{} +} + +func cgrRPCAction(account *Account, sq *StatsQueueTriggered, a *Action, acs Actions) error { + req := RPCRequest{} + if err := json.Unmarshal([]byte(a.ExtraParameters), &req); err != nil { + return err + } + params, err := utils.GetRpcParams(req.Method) + if err != nil { + return err + } + var client rpcclient.RpcClientConnection + if req.Address != utils.INTERNAL { + if client, err = rpcclient.NewRpcClient(req.Method, req.Address, req.Attempts, 0, req.Transport, nil); err != nil { + return err + } + } else { + client = params.Object + } + if client == nil { + return utils.ErrServerError + } + + in, out := params.InParam, params.OutParam + p, err := utils.FromMapStringInterfaceValue(req.Param, in) + if err != nil { + return err + } + if !req.Async { + err = client.Call(req.Method, p, out) + utils.Logger.Info(fmt.Sprintf("<*cgr_rpc> result: %+v err: %v", out, err)) + return err + } + go func() { + err := client.Call(req.Method, p, out) + utils.Logger.Info(fmt.Sprintf("<*cgr_rpc> result: %+v err: %v", out, err)) + }() + return nil +} + // Structure to store actions according to weight type Actions []*Action diff --git a/engine/actions_test.go b/engine/actions_test.go index bead47620..f9297d25a 100644 --- a/engine/actions_test.go +++ b/engine/actions_test.go @@ -22,6 +22,7 @@ import ( "encoding/json" "fmt" "reflect" + "strings" "testing" "time" @@ -2102,6 +2103,8 @@ func TestActionCdrlogBalanceValue(t *testing.T) { ID: "cgrates.org:bv", BalanceMap: map[string]Balances{ utils.MONETARY: Balances{&Balance{ + ID: "*default", + Uuid: "25a02c82-f09f-4c6e-bacf-8ed4b076475a", Value: 10, }}, }, @@ -2114,16 +2117,29 @@ func TestActionCdrlogBalanceValue(t *testing.T) { Timing: &RateInterval{}, actions: []*Action{ &Action{ + Id: "RECUR_FOR_V3HSILLMILLD1G", ActionType: TOPUP, - Balance: &BalanceFilter{Value: utils.Float64Pointer(1.1), Type: utils.StringPointer(utils.MONETARY)}, + Balance: &BalanceFilter{ + ID: utils.StringPointer("*default"), + Uuid: utils.StringPointer("25a02c82-f09f-4c6e-bacf-8ed4b076475a"), + Value: utils.Float64Pointer(1.1), + Type: utils.StringPointer(utils.MONETARY), + }, }, &Action{ + Id: "RECUR_FOR_V3HSILLMILLD5G", ActionType: DEBIT, - Balance: &BalanceFilter{Value: utils.Float64Pointer(2.1), Type: utils.StringPointer(utils.MONETARY)}, + Balance: &BalanceFilter{ + ID: utils.StringPointer("*default"), + Uuid: utils.StringPointer("25a02c82-f09f-4c6e-bacf-8ed4b076475a"), + Value: utils.Float64Pointer(2.1), + Type: utils.StringPointer(utils.MONETARY), + }, }, &Action{ + Id: "c", ActionType: CDRLOG, - ExtraParameters: `{"BalanceValue":"BalanceValue"}`, + ExtraParameters: `{"BalanceID":"BalanceID","BalanceUUID":"BalanceUUID","ActionID":"ActionID","BalanceValue":"BalanceValue"}`, }, }, } @@ -2140,7 +2156,69 @@ func TestActionCdrlogBalanceValue(t *testing.T) { if len(cdrs) != 2 || cdrs[0].ExtraFields["BalanceValue"] != "11.1" || cdrs[1].ExtraFields["BalanceValue"] != "9" { - t.Errorf("Wrong cdrlogs: %+v", cdrs[1]) + t.Errorf("Wrong cdrlogs: %", utils.ToIJSON(cdrs)) + } +} + +type TestRPCParameters struct { + status string +} + +type Attr struct { + Name string + Surname string + Age float64 +} + +func (trpcp *TestRPCParameters) Hopa(in Attr, out *float64) error { + trpcp.status = utils.OK + return nil +} + +func (trpcp *TestRPCParameters) Call(serviceMethod string, args interface{}, reply interface{}) error { + parts := strings.Split(serviceMethod, ".") + if len(parts) != 2 { + return utils.ErrNotImplemented + } + // get method + method := reflect.ValueOf(trpcp).MethodByName(parts[1]) + if !method.IsValid() { + return utils.ErrNotImplemented + } + + // construct the params + params := []reflect.Value{reflect.ValueOf(args), reflect.ValueOf(reply)} + + ret := method.Call(params) + if len(ret) != 1 { + return utils.ErrServerError + } + if ret[0].Interface() == nil { + return nil + } + err, ok := ret[0].Interface().(error) + if !ok { + return utils.ErrServerError + } + return err +} + +func TestCgrRpcAction(t *testing.T) { + trpcp := &TestRPCParameters{} + utils.RegisterRpcParams("", trpcp) + a := &Action{ + ExtraParameters: `{"Address": "internal", + "Transport": "*gob", + "Method": "TestRPCParameters.Hopa", + "Attempts":1, + "Async" :false, + "Param": {"Name":"n", "Surname":"s", "Age":10.2}}`, + } + if err := cgrRPCAction(nil, nil, a, nil); err != nil { + t.Error("error executing cgr action: ", err) + } + if trpcp.status != utils.OK { + t.Error("RPC not called!") } } diff --git a/general_tests/tp_it_test.go b/general_tests/tp_it_test.go index cb66dca3e..8165ed19c 100644 --- a/general_tests/tp_it_test.go +++ b/general_tests/tp_it_test.go @@ -214,7 +214,7 @@ func TestTpZeroNegativeCost(t *testing.T) { } else if cc.GetDuration() != 20*time.Second { t.Errorf("Calling Responder.MaxDebit got callcost: %v", utils.ToIJSON(cc)) } - var acnt *engine.Account + var acnt engine.Account attrs := &utils.AttrGetAccount{Tenant: "cgrates.org", Account: "1013"} if err := tpRPC.Call("ApierV2.GetAccount", attrs, &acnt); err != nil { t.Error("Got error on ApierV2.GetAccount: ", err.Error()) @@ -222,3 +222,20 @@ func TestTpZeroNegativeCost(t *testing.T) { t.Errorf("Calling ApierV2.GetAccount received: %s", utils.ToIJSON(acnt)) } } + +func TestTpExecuteActionCgrRpc(t *testing.T) { + if !*testIntegration { + return + } + var reply string + if err := tpRPC.Call("ApierV2.ExecuteAction", utils.AttrExecuteAction{ActionsId: "RPC"}, &reply); err != nil { + t.Error("Got error on ApierV2.ExecuteAction: ", err.Error()) + } else if reply != utils.OK { + t.Errorf("Calling ExecuteAction got reply: %s", reply) + } + var acnt engine.Account + attrs := &utils.AttrGetAccount{Tenant: "cgrates.org", Account: "rpc"} + if err := tpRPC.Call("ApierV2.GetAccount", attrs, &acnt); err != nil { + t.Error("Got error on ApierV2.GetAccount: ", err.Error()) + } +} diff --git a/utils/rpc_params.go b/utils/rpc_params.go new file mode 100644 index 000000000..69020fdec --- /dev/null +++ b/utils/rpc_params.go @@ -0,0 +1,50 @@ +package utils + +import ( + "reflect" + + "github.com/cgrates/rpcclient" +) + +var rpcParamsMap map[string]*RpcParams + +type RpcParams struct { + Object rpcclient.RpcClientConnection + InParam reflect.Value + OutParam interface{} +} + +func init() { + rpcParamsMap = make(map[string]*RpcParams) +} + +func RegisterRpcParams(name string, obj rpcclient.RpcClientConnection) { + objType := reflect.TypeOf(obj) + if name == "" { + val := reflect.ValueOf(obj) + name = objType.Name() + if val.Kind() == reflect.Ptr { + name = objType.Elem().Name() + } + } + for i := 0; i < objType.NumMethod(); i++ { + method := objType.Method(i) + methodType := method.Type + if methodType.NumIn() == 3 { // if it has three parameters (one is self and two are rpc params) + rpcParamsMap[name+"."+method.Name] = &RpcParams{ + Object: obj, + InParam: reflect.New(methodType.In(1)), + OutParam: reflect.New(methodType.In(2).Elem()).Interface(), + } + } + + } +} + +func GetRpcParams(method string) (*RpcParams, error) { + x, found := rpcParamsMap[method] + if !found { + return nil, ErrNotFound + } + return x, nil +} diff --git a/utils/rpc_params_test.go b/utils/rpc_params_test.go new file mode 100644 index 000000000..aac04ecdc --- /dev/null +++ b/utils/rpc_params_test.go @@ -0,0 +1,48 @@ +package utils + +import "testing" + +type RpcStruct struct{} + +type Attr struct { + Name string + Surname string + Age float64 +} + +func (rpc *RpcStruct) Hopa(normal Attr, out *float64) error { + return nil +} + +func (rpc *RpcStruct) Tropa(pointer *Attr, out *float64) error { + return nil +} + +func (rpc *RpcStruct) Call(string, interface{}, interface{}) error { + return nil +} + +func TestRPCObjectPointer(t *testing.T) { + RegisterRpcParams("", &RpcStruct{}) + if len(rpcParamsMap) != 2 { + t.Errorf("error registering rpc object: %v", rpcParamsMap) + } + x, found := rpcParamsMap["RpcStruct.Hopa"] + if !found { + t.Errorf("error getting rpcobject: %v (%+v)", rpcParamsMap, x) + } + a := x.InParam + if v, err := FromMapStringInterfaceValue(map[string]interface{}{"Name": "a", "Surname": "b", "Age": 10.2}, a); err != nil || v.(Attr).Name != "a" || v.(Attr).Surname != "b" || v.(Attr).Age != 10.2 { + t.Errorf("error converting to struct: %+v (%v)", v, err) + } + //TODO: make pointer in arguments usable + /*x, found = rpcParamsMap["RpcStruct.Tropa"] + if !found { + t.Errorf("error getting rpcobject: %v (%+v)", rpcParamsMap, x) + } + b := x.InParam + log.Printf("T: %+v", b) + if v, err := FromMapStringInterfaceValue(map[string]interface{}{"Name": "a", "Surname": "b", "Age": 10.2}, b); err != nil || v.(Attr).Name != "a" || v.(Attr).Surname != "b" || v.(Attr).Age != 10.2 { + t.Errorf("error converting to struct: %+v (%v)", v, err) + }*/ +} diff --git a/utils/struct.go b/utils/struct.go index 0b16caeef..e6f20fff3 100644 --- a/utils/struct.go +++ b/utils/struct.go @@ -18,6 +18,7 @@ along with this program. If not, see package utils import ( + "errors" "reflect" "strconv" "strings" @@ -171,6 +172,49 @@ func FromMapStringString(m map[string]string, in interface{}) { return } +func FromMapStringInterface(m map[string]interface{}, in interface{}) error { + v := reflect.ValueOf(in) + if v.Kind() == reflect.Ptr { + v = v.Elem() + } + for fieldName, fieldValue := range m { + field := v.FieldByName(fieldName) + if field.IsValid() { + if !field.IsValid() || !field.CanSet() { + continue + } + structFieldType := field.Type() + val := reflect.ValueOf(fieldValue) + if structFieldType != val.Type() { + return errors.New("Provided value type didn't match obj field type") + } + field.Set(val) + } + } + return nil +} + +func FromMapStringInterfaceValue(m map[string]interface{}, v reflect.Value) (interface{}, error) { + if v.Kind() == reflect.Ptr { + v = v.Elem() + } + for fieldName, fieldValue := range m { + field := v.FieldByName(fieldName) + if field.IsValid() { + if !field.IsValid() || !field.CanSet() { + continue + } + structFieldType := field.Type() + val := reflect.ValueOf(fieldValue) + if structFieldType != val.Type() { + return nil, errors.New("Provided value type didn't match obj field type") + } + field.Set(val) + } + } + return v.Interface(), nil +} + // Update struct with map fields, returns not matching map keys, s is a struct to be updated func UpdateStructWithStrMap(s interface{}, m map[string]string) []string { notMatched := []string{} diff --git a/utils/struct_test.go b/utils/struct_test.go index 50371ea36..74450a7cb 100644 --- a/utils/struct_test.go +++ b/utils/struct_test.go @@ -84,3 +84,32 @@ func TestStructExtraFields(t *testing.T) { t.Errorf("expected: %v got: %v", ts.ExtraFields, efMap) } } + +func TestStructFromMapStringInterface(t *testing.T) { + ts := &struct { + Name string + Class *string + List []string + Elements struct { + Type string + Value float64 + } + }{} + s := "test2" + m := map[string]interface{}{ + "Name": "test1", + "Class": &s, + "List": []string{"test3", "test4"}, + "Elements": struct { + Type string + Value float64 + }{ + Type: "test5", + Value: 9.8, + }, + } + if err := FromMapStringInterface(m, ts); err != nil { + t.Logf("ts: %+v", ToJSON(ts)) + t.Error("Error converting map to struct: ", err) + } +}