better cgr_rpc and more tests

data/tariffplans/testtp/AccountActions.csv

@@ -11,3 +11,4 @@ cgrates.org,1012,PREPAID_10,,,
 cgrates.org,1013,TEST_NEG,,,
 cgrates.org,1014,TEST_RPC,,,
 cgrates.org,1015,TEST_DID,,,
+cgrates.org,1016,PREPAID_10,,,

data/tariffplans/testtp/Actions.csv

@@ -12,3 +12,5 @@ TOPUP_NEG,*topup,,,,*voice,*out,,GERMANY;!GERMANY_MOBILE,*zero1m,,*unlimited,,10
 RPC,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""ApierV2.SetAccount"",""Attempts"":1,""Async"" :false,""Params"":{""Account"":""rpc"",""Tenant"":""cgrates.org""}}",,,,,,,,,,,,,,,
 DID,*debit,,,,*monetary,*out,,*any,,,*unlimited,*any,"{""Method"":""*incremental"",""Params"":{""Units"":1, ""Interval"":""month"",""Increment"":""day""}}",10.0,,,10.0
 DID,*cdrlog,"{""action"":""^DID"",""prev_balance"":""BalanceValue""}",,,*monetary,*out,,*any,,,*unlimited,,,10.0,,,10.0
+RPC_DEST,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""ApierV2.SetDestination"",""Attempts"":1,""Async"" :false,""Params"":{""Id"":""<<.Account.GetID>>"",""Prefixes"":[""1"",""2"",""3""]}}",,,,,,,,,,,,,,,
+RPC_CDRSTATS,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""CDRStatsV1.AddQueue"",""Attempts"":1,""Async"" :false,""Params"":{""Id"":""qtest""}}",,,,,,,,,,,,,,,

engine/account.go

@@ -843,6 +843,14 @@ func (acc *Account) matchActionFilter(condition string) (bool, error) {
 	return false, nil
 }
 
+func (acc *Account) GetID() string {
+	split := strings.Split(acc.ID, utils.CONCATENATED_KEY_SEP)
+	if len(split) != 2 {
+		return ""
+	}
+	return split[1]
+}
+
 // used in some api for transition
 func (acc *Account) AsOldStructure() interface{} {
 	type Balance struct {

engine/action.go

@@ -35,6 +35,7 @@ import (
 	"github.com/cgrates/cgrates/config"
 	"github.com/cgrates/cgrates/utils"
 	"github.com/cgrates/rpcclient"
+	"github.com/mitchellh/mapstructure"
 )
 
 /*
@@ -642,22 +643,24 @@ type RPCRequest struct {
 func cgrRPCAction(account *Account, sq *StatsQueueTriggered, a *Action, acs Actions) error {
 	// parse template
 	tmpl := template.New("extra_params")
+	tmpl.Delims("<<", ">>")
 	t, err := tmpl.Parse(a.ExtraParameters)
 	if err != nil {
 		utils.Logger.Err(fmt.Sprintf("error parsing *cgr_rpc template: %s", err.Error()))
 		return err
 	}
 	var buf bytes.Buffer
-	if err = t.Execute(&buf, map[string]interface{}{
-		"account": account,
-		"action":  a,
-		"actions": acs,
-		"sq":      sq,
-	}); err != nil {
+	if err = t.Execute(&buf, struct {
+		Account *Account
+		Sq      *StatsQueueTriggered
+		Action  *Action
+		Actions Actions
+	}{account, sq, a, acs}); err != nil {
 		utils.Logger.Err(fmt.Sprintf("error executing *cgr_rpc template %s:", err.Error()))
 		return err
 	}
 	a.ExtraParameters = buf.String()
+	//utils.Logger.Info("ExtraParameters: " + a.ExtraParameters)
 	req := RPCRequest{}
 	if err := json.Unmarshal([]byte(a.ExtraParameters), &req); err != nil {
 		return err
@@ -675,17 +678,21 @@ func cgrRPCAction(account *Account, sq *StatsQueueTriggered, a *Action, acs Acti
 		client = params.Object.(rpcclient.RpcClientConnection)
 	}
 	in, out := params.InParam, params.OutParam
-	p, err := utils.FromMapStringInterfaceValue(req.Params, in)
+	//utils.Logger.Info("Params: " + utils.ToJSON(req.Params))
+	//p, err := utils.FromMapStringInterfaceValue(req.Params, in)
+	mapstructure.Decode(req.Params, in)
 	if err != nil {
+		utils.Logger.Info("err3: " + err.Error())
 		return err
 	}
+	utils.Logger.Info(fmt.Sprintf("<*cgr_rpc> calling: %s with: %s", req.Method, utils.ToJSON(in)))
 	if !req.Async {
-		err = client.Call(req.Method, p, out)
+		err = client.Call(req.Method, in, out)
 		utils.Logger.Info(fmt.Sprintf("<*cgr_rpc> result: %s err: %v", utils.ToJSON(out), err))
 		return err
 	}
 	go func() {
-		err := client.Call(req.Method, p, out)
+		err := client.Call(req.Method, in, out)
 		utils.Logger.Info(fmt.Sprintf("<*cgr_rpc> result: %s err: %v", utils.ToJSON(out), err))
 	}()
 	return nil

engine/action_trigger.go

@@ -30,7 +30,7 @@ import (
 type ActionTrigger struct {
 	ID            string // original csv tag
 	UniqueID      string // individual id
-	ThresholdType string //*min_event_counter, *max_event_counter, *min_balance_counter, *max_balance_counter, *min_balance, *max_balance, *exp_balance
+	ThresholdType string //*min_event_counter, *max_event_counter, *min_balance_counter, *max_balance_counter, *min_balance, *max_balance, *balance_expired
 	// stats: *min_asr, *max_asr, *min_acd, *max_acd, *min_tcd, *max_tcd, *min_acc, *max_acc, *min_tcc, *max_tcc, *min_ddc, *max_ddc
 	ThresholdValue float64
 	Recurrent      bool          // reset excuted flag each run

general_tests/tp_it_test.go

@@ -241,6 +241,45 @@ func TestTpExecuteActionCgrRpc(t *testing.T) {
 	}
 }
 
+func TestTpExecuteActionCgrRpcAcc(t *testing.T) {
+	if !*testIntegration {
+		return
+	}
+	var reply string
+	if err := tpRPC.Call("ApierV2.ExecuteAction", utils.AttrExecuteAction{
+		Tenant:    "cgrates.org",
+		Account:   "1016",
+		ActionsId: "RPC_DEST",
+	}, &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 dests []*engine.Destination
+	attrs := &v2.AttrGetDestinations{DestinationIDs: []string{"1016"}}
+	if err := tpRPC.Call("ApierV2.GetDestinations", attrs, &dests); err != nil {
+		t.Error("Got error on ApierV2.GetDestinations: ", err.Error())
+	}
+}
+
+func TestTpExecuteActionCgrRpcCdrStats(t *testing.T) {
+	if !*testIntegration {
+		return
+	}
+	var reply string
+	if err := tpRPC.Call("ApierV2.ExecuteAction", utils.AttrExecuteAction{
+		ActionsId: "RPC_CDRSTATS",
+	}, &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 queue engine.StatsQueue
+	if err := tpRPC.Call("CDRStatsV1.GetQueue", "qtest", &queue); err != nil {
+		t.Error("Got error on CDRStatsV1.GetQueue: ", err.Error())
+	}
+}
+
 func TestTpCreateExecuteActionMatch(t *testing.T) {
 	if !*testIntegration {
 		return

glide.yaml

@@ -35,3 +35,4 @@ import:
   subpackages:
   - bson
 - package: github.com/ChrisTrenkamp/goxpath
+- package: github.com/mitchellh/mapstructure

utils/rpc_params.go

@@ -6,7 +6,7 @@ var rpcParamsMap map[string]*RpcParams
 
 type RpcParams struct {
 	Object   interface{}
-	InParam  reflect.Value
+	InParam  interface{}
 	OutParam interface{}
 }
 
@@ -33,7 +33,7 @@ func RegisterRpcParams(name string, obj interface{}) {
 			}
 			rpcParamsMap[name+"."+method.Name] = &RpcParams{
 				Object:   obj,
-				InParam:  reflect.New(methodType.In(1)),
+				InParam:  reflect.New(methodType.In(1)).Interface(),
 				OutParam: reflect.New(out).Interface(),
 			}
 		}

utils/struct.go

@@ -19,6 +19,7 @@ package utils
 
 import (
 	"errors"
+	"fmt"
 	"reflect"
 	"strconv"
 	"strings"
@@ -194,6 +195,7 @@ func FromMapStringInterface(m map[string]interface{}, in interface{}) error {
 	return nil
 }
 
+// initial intent was to use it with *cgr_rpc but does not handle slice and structure fields
 func FromMapStringInterfaceValue(m map[string]interface{}, v reflect.Value) (interface{}, error) {
 	if v.Kind() == reflect.Ptr {
 		v = v.Elem()
@@ -204,10 +206,15 @@ func FromMapStringInterfaceValue(m map[string]interface{}, v reflect.Value) (int
 			if !field.IsValid() || !field.CanSet() {
 				continue
 			}
-			structFieldType := field.Type()
 			val := reflect.ValueOf(fieldValue)
+			structFieldType := field.Type()
+			if structFieldType.Kind() == reflect.Ptr {
+				field.Set(reflect.New(field.Type().Elem()))
+				field = field.Elem()
+			}
+			structFieldType = field.Type()
 			if structFieldType != val.Type() {
-				return nil, errors.New("Provided value type didn't match obj field type")
+				return nil, fmt.Errorf("provided value type didn't match obj field type: %v vs %v (%v vs %v)", structFieldType, val.Type(), structFieldType.Kind(), val.Type().Kind())
 			}
 			field.Set(val)
 		}

utils/struct_test.go

@@ -1,6 +1,7 @@
 package utils
 
 import (
+	"cmd/cgrates/utils"
 	"reflect"
 	"testing"
 )
@@ -113,3 +114,27 @@ func TestStructFromMapStringInterface(t *testing.T) {
 		t.Error("Error converting map to struct: ", err)
 	}
 }
+
+func TestStructFromMapStringInterfaceValue(t *testing.T) {
+	type T struct {
+		Name     string
+		Disabled *bool
+		Members  []string
+	}
+	ts := &T{}
+	vts := reflect.ValueOf(ts)
+	x, err := FromMapStringInterfaceValue(map[string]interface{}{
+		"Name":     "test",
+		"Disabled": true,
+		"Members":  []string{"1", "2", "3"},
+	}, vts)
+	rt := x.(T)
+	if err != nil {
+		t.Fatalf("error converting structure value: %v", err)
+	}
+	if rt.Name != "test" ||
+		*rt.Disabled != true ||
+		!reflect.DeepEqual(rt.Members, []string{"1", "2", "3"}) {
+		t.Errorf("error converting structure value: %s", utils.ToIJSON(rt))
+	}
+}