Merge branch 'master' into hapool

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

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)
 		}

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""}}",,,,,,,,,,,,,,,

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
 

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!")
 	}
 }
 

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())
+	}
+}

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
+}

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)
+	}*/
+}

utils/struct.go

@@ -18,6 +18,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 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{}

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)
+	}
+}