Added tests and fixes for DispatcherService.AttributeSv1

apier/v1/apier_it_test.go

@@ -1944,7 +1944,7 @@ func TestApierPing(t *testing.T) {
 	var reply string
 	for _, method := range []string{utils.StatSv1Ping, utils.ResourceSv1Ping,
 		utils.SupplierSv1Ping, utils.ThresholdSv1Ping, utils.AttributeSv1Ping} {
-		if err := rater.Call(method, "", &reply); err != nil {
+		if err := rater.Call(method, utils.CGREvent{}, &reply); err != nil {
 			t.Error(err)
 		} else if reply != utils.Pong {
 			t.Errorf("Received: %s", reply)

apier/v1/attributes_it_test.go

@@ -847,7 +847,7 @@ func testAttributeSSetAlsPrf4(t *testing.T) {
 
 func testAttributeSPing(t *testing.T) {
 	var resp string
-	if err := attrSRPC.Call(utils.AttributeSv1Ping, "", &resp); err != nil {
+	if err := attrSRPC.Call(utils.AttributeSv1Ping, utils.CGREvent{}, &resp); err != nil {
 		t.Error(err)
 	} else if resp != utils.Pong {
 		t.Error("Unexpected reply returned", resp)

apier/v1/chargers.go

@@ -97,7 +97,7 @@ func (cSv1 *ChargerSv1) Call(serviceMethod string,
 	return utils.APIerRPCCall(cSv1, serviceMethod, args, reply)
 }
 
-func (cSv1 *ChargerSv1) Ping(ign string, reply *string) error {
+func (cSv1 *ChargerSv1) Ping(ign utils.CGREvent, reply *string) error {
 	*reply = utils.Pong
 	return nil
 }

apier/v1/dispatcher.go

@@ -190,12 +190,6 @@ type DispatcherAttributeSv1 struct {
 	dA *dispatchers.DispatcherService
 }
 
-// Call implements rpcclient.RpcClientConnection interface for internal RPC
-// func (alSv1 *DispatcherAttributeSv1) Call(serviceMethod string,
-// 	args interface{}, reply interface{}) error {
-// 	return utils.APIerRPCCall(alSv1, serviceMethod, args, reply)
-// }
-
 // Ping implements SupplierSv1Ping
 func (dA *DispatcherAttributeSv1) Ping(args *dispatchers.CGREvWithApiKey, reply *string) error {
 	return dA.dA.AttributeSv1Ping(args, reply)
@@ -263,7 +257,7 @@ func (dS *DispatcherSessionSv1) UpdateSession(args *dispatchers.UpdateSessionWit
 	reply *sessions.V1UpdateSessionReply) (err error) {
 	return dS.dS.SessionSv1UpdateSession(args, reply)
 }
-
+*/
 func NewDispatcherChargerSv1(dps *dispatchers.DispatcherService) *DispatcherChargerSv1 {
 	return &DispatcherChargerSv1{dC: dps}
 }
@@ -274,8 +268,8 @@ type DispatcherChargerSv1 struct {
 }
 
 // Ping implements ChargerSv1Ping
-func (dC *DispatcherChargerSv1) Ping(ign string, reply *string) error {
-	return dC.dC.ChargerSv1Ping(ign, reply)
+func (dC *DispatcherChargerSv1) Ping(args *dispatchers.CGREvWithApiKey, reply *string) error {
+	return dC.dC.ChargerSv1Ping(args, reply)
 }
 
 // GetChargersForEvent implements ChargerSv1GetChargersForEvent
@@ -289,4 +283,3 @@ func (dC *DispatcherChargerSv1) ProcessEvent(args *dispatchers.CGREvWithApiKey,
 	reply *[]*engine.AttrSProcessEventReply) (err error) {
 	return dC.dC.ChargerSv1ProcessEvent(args, reply)
 }
-*/

cmd/cgr-engine/cgr-engine.go

@@ -1042,20 +1042,17 @@ func startDispatcherService(internalDispatcherSChan chan *dispatchers.Dispatcher
 				v1.NewDispatcherSupplierSv1(dspS))
 		}
 	*/
-	// if !cfg.AttributeSCfg().Enabled { //dispatcer enable all methos
-	attrv1 := v1.NewDispatcherAttributeSv1(dspS)
-	server.RpcRegisterName(utils.AttributeSv1, attrv1)
-	// }
+	server.RpcRegisterName(utils.AttributeSv1,
+		v1.NewDispatcherAttributeSv1(dspS))
 	/*
 		if !cfg.SessionSCfg().Enabled && len(cfg.DispatcherSCfg().SessionSConns) != 0 {
 			server.RpcRegisterName(utils.SessionSv1,
 				v1.NewDispatcherSessionSv1(dspS))
 		}
-		if !cfg.ChargerSCfg().Enabled && len(cfg.DispatcherSCfg().ChargerSConns) != 0 {
-			server.RpcRegisterName(utils.ChargerSv1,
-				v1.NewDispatcherChargerSv1(dspS))
-		}
 	*/
+	server.RpcRegisterName(utils.ChargerSv1,
+		v1.NewDispatcherChargerSv1(dspS))
+
 	internalDispatcherSChan <- dspS
 }
 

console/ping.go

@@ -19,8 +19,9 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package console
 
 import (
-	"github.com/cgrates/cgrates/utils"
 	"strings"
+
+	"github.com/cgrates/cgrates/utils"
 )
 
 func init() {
@@ -79,7 +80,7 @@ func (self *CmdApierPing) RpcParams(reset bool) interface{} {
 		self.rpcParams = &StringWrapper{}
 	}
 
-	return self.rpcParams
+	return utils.CGREvent{}
 }
 
 func (self *CmdApierPing) PostprocessRpcParams() error {

data/conf/samples/dispatchers/all/cgrates.json

@@ -0,0 +1,43 @@
+{
+// CGRateS Configuration file
+//
+
+
+"general": {
+	"node_id": "ALL",
+	"log_level": 7
+},
+
+
+"listen": {
+ 	"rpc_json": ":6012",
+	"rpc_gob": ":6013",
+	"http": ":6080",
+},
+
+"data_db": {								// database used to store runtime data (eg: accounts, cdr stats)
+	"db_type": "redis",						// data_db type: <redis|mongo>
+	"db_port": 6379, 						// data_db port to reach the database
+	"db_name": "11", 						// data_db database name to connect to
+},
+
+"stor_db": {
+	"db_type":"*internal",
+},
+
+"attributes": {
+	"enabled": true
+},
+
+"rals": {
+	"enabled": true,
+},
+
+// "chargers": {
+// 	"enabled": true,
+// 	"attributes_conns": [
+// 		{"address": "*internal"},
+// 	],
+// },
+
+}

data/conf/samples/dispatchers/all2/cgrates.json

@@ -0,0 +1,43 @@
+{
+// CGRateS Configuration file
+//
+
+
+"general": {
+	"node_id": "ALL2",
+	"log_level": 7
+},
+
+
+"listen": {
+ 	"rpc_json": ":7012",
+	"rpc_gob": ":7013",
+	"http": ":7080",
+},
+
+"data_db": {								// database used to store runtime data (eg: accounts, cdr stats)
+	"db_type": "redis",						// data_db type: <redis|mongo>
+	"db_port": 6379, 						// data_db port to reach the database
+	"db_name": "12", 						// data_db database name to connect to
+},
+
+"stor_db": {
+	"db_type":"*internal",
+},
+
+"attributes": {
+	"enabled": true
+},
+
+"rals": {
+	"enabled": true,
+},
+
+// "chargers": {
+// 	"enabled": true,
+// 	"attributes_conns": [
+// 		{"address": "*internal"},
+// 	],
+// },
+
+}

data/conf/samples/dispatchers/attributes/cgrates.json

@@ -8,31 +8,22 @@
 	"log_level": 7
 },
 
-
 "listen": {
  	"rpc_json": ":5012",
 	"rpc_gob": ":5013",
 	"http": ":5080",
 },
 
-"data_db": {								// database used to store runtime data (eg: accounts, cdr stats)
-	"db_type": "redis",						// data_db type: <redis|mongo>
-	"db_port": 6379, 						// data_db port to reach the database
-	"db_name": "10", 						// data_db database name to connect to
-},
-
 "stor_db": {
-	"db_password": "CGRateS.org",
+	"db_type":"*internal",
 },
 
-
 "attributes": {
 	"enabled": true
 },
 
-
 "rals": {
 	"enabled": true,
 },
-
 }
+ 

data/conf/samples/dispatchers/dispatchers/cgrates.json

@@ -7,7 +7,8 @@
 // This is what you get when you load CGRateS with an empty configuration file.
 
 "general": {
- 	"node_id": "DispatcherS1"
+ 	"node_id": "DispatcherS1",
+ 	"reconnects": 1,	
 },
 
 
@@ -17,14 +18,8 @@
 	"http": ":2080",
 },
 
-"data_db": {								// database used to store runtime data (eg: accounts, cdr stats)
-	"db_type": "redis",						// data_db type: <redis|mongo>
-	"db_port": 6379, 						// data_db port to reach the database
-	"db_name": "10", 						// data_db database name to connect to
-},
-
 "stor_db": {
-	"db_password": "CGRateS.org",
+	"db_type":"*internal",
 },
 
 "cache":{
@@ -38,18 +33,20 @@
 
 
 "dispatchers":{
- 	"enabled": true,						// starts DispatcherS service: <true|false>.
+ 	"enabled": true,
  	"attributes_conns": [
 		{"address": "127.0.0.1:5012", "transport": "*json"},
- 		// {"address": "*internal"}
  	],
 	"conns": {
-		// "AttributeS1": [
-		// 	{"address": "127.0.0.1:2012", "transport": "*json"},
-		// ],
-		"ALL": [
+		"AttributeS1": [
 			{"address": "127.0.0.1:5012", "transport": "*json"},
 		],
+		"ALL": [
+			{"address": "127.0.0.1:6012", "transport": "*json"},
+		],
+		"ALL2": [
+			{"address": "127.0.0.1:7012", "transport": "*json"},
+		],
 	},
 },
 

data/tariffplans/dispatchers/Attributes.csv

@@ -0,0 +1,6 @@
+#Tenant,ID,Contexts,FilterIDs,ActivationInterval,FieldName,Initial,Substitute,Append,Blocker,Weight
+cgrates.org,ATTR_1001_SIMPLEAUTH,simpleauth,*string:Account:1001,,Password,*any,CGRateS.org,true,false,20
+cgrates.org,ATTR_API_ATTR_FAKE_AUTH,*auth,*string:APIKey:12345,,APIMethods,*any,,true,false,20
+cgrates.org,ATTR_API_ATTR_AUTH,*auth,*string:APIKey:attr12345,,APIMethods,*any,AttributeSv1.Ping&AttributeSv1.GetAttributeForEvent&AttributeSv1.ProcessEvent,true,false,20
+
+

data/tariffplans/dispatchers/Dispatchers.csv

@@ -0,0 +1,5 @@
+#Tenant,ID,Contexts,FilterIDs,ActivationInterval,Strategy,StrategyParameters,ConnID,ConnFilterIDs,ConnWeight,ConnBlocker,ConnParameters,Weight
+cgrates.org,PING1,*any,,,*weight,,ALL,,20,false,,10
+cgrates.org,PING1,,,,,,ALL2,,10,,,
+cgrates.org,EVENT1,*any,*string:EventName:Event1,,*weight,,ALL2,,20,false,,20
+cgrates.org,EVENT1,,,,,,ALL,,10,,,

dispatchers/attributes.go

@@ -49,7 +49,6 @@ func (dS *DispatcherService) AttributeSv1GetAttributeForEvent(args *ArgsAttrProc
 	}
 	return dS.Dispatch(&args.CGREvent, utils.MetaAttributes,
 		utils.AttributeSv1GetAttributeForEvent, args.AttrArgsProcessEvent, reply)
-
 }
 
 func (dS *DispatcherService) AttributeSv1ProcessEvent(args *ArgsAttrProcessEventWithApiKey,

dispatchers/attributes_it_test.go

@@ -23,8 +23,10 @@ package dispatchers
 import (
 	"net/rpc"
 	"net/rpc/jsonrpc"
+	"os/exec"
 	"path"
 	"reflect"
+	"strconv"
 	"testing"
 	"time"
 
@@ -33,112 +35,260 @@ import (
 	"github.com/cgrates/cgrates/utils"
 )
 
+type testDipatcer struct {
+	CfgParh string
+	Cfg     *config.CGRConfig
+	RCP     *rpc.Client
+	cmd     *exec.Cmd
+}
+
+func newTestEngine(t *testing.T, cfgPath string, initDataDB, intitStoreDB bool) (d *testDipatcer) {
+	d = new(testDipatcer)
+	d.CfgParh = cfgPath
+	var err error
+	d.Cfg, err = config.NewCGRConfigFromFolder(d.CfgParh)
+	if err != nil {
+		t.Fatalf("Error at config init :%v\n", err)
+	}
+	d.Cfg.DataFolderPath = dspDataDir // Share DataFolderPath through config towards StoreDb for Flush()
+
+	if initDataDB {
+		d.initDataDb(t)
+	}
+
+	if intitStoreDB {
+		d.resetStorDb(t)
+	}
+	d.startEngine(t)
+	return d
+}
+
+func (d *testDipatcer) startEngine(t *testing.T) {
+	var err error
+	if d.cmd, err = engine.StartEngine(d.CfgParh, dspDelay); err != nil {
+		t.Fatalf("Error at engine start:%v\n", err)
+	}
+
+	if d.RCP, err = jsonrpc.Dial("tcp", d.Cfg.ListenCfg().RPCJSONListen); err != nil {
+		t.Fatalf("Error at dialing rcp client:%v\n", err)
+	}
+}
+
+func (d *testDipatcer) stopEngine(t *testing.T) {
+	pid := strconv.Itoa(d.cmd.Process.Pid)
+	if err := exec.Command("kill", "-9", pid).Run(); err != nil {
+		t.Fatalf("Error at stop engine:%v\n", err)
+	}
+	// // if err := d.cmd.Process.Kill(); err != nil {
+	// // 	t.Fatalf("Error at stop engine:%v\n", err)
+	// }
+}
+
+func (d *testDipatcer) initDataDb(t *testing.T) {
+	if err := engine.InitDataDb(d.Cfg); err != nil {
+		t.Fatalf("Error at DataDB init:%v\n", err)
+	}
+}
+
+// Wipe out the cdr database
+func (d *testDipatcer) resetStorDb(t *testing.T) {
+	if err := engine.InitStorDb(d.Cfg); err != nil {
+		t.Fatalf("Error at DataDB init:%v\n", err)
+	}
+}
+func (d *testDipatcer) loadData(t *testing.T, path string) {
+	var reply string
+	attrs := &utils.AttrLoadTpFromFolder{FolderPath: path}
+	if err := d.RCP.Call("ApierV1.LoadTariffPlanFromFolder", attrs, &reply); err != nil {
+		t.Errorf("Error at loading data from folder:%v", err)
+	}
+}
+
 var (
-	dspAttrCfgPath  string
-	dspAttrCfg      *config.CGRConfig
-	dspAttrRPC      *rpc.Client
-	instAttrCfgPath string
-	instAttrCfg     *config.CGRConfig
-	instAttrRPC     *rpc.Client
+	attrEngine *testDipatcer
+	dispEngine *testDipatcer
+	allEngine  *testDipatcer
+	allEngine2 *testDipatcer
 )
 
 var sTestsDspAttr = []func(t *testing.T){
-	testDspAttrInitCfg,
-	testDspAttrInitDataDb,
-	testDspAttrResetStorDb,
-	testDspAttrStartEngine,
-	testDspAttrRPCConn,
-	testDspAttrLoadData,
+	testDspAttrPingFailover,
+	testDspAttrGetAttrFailover,
+
 	testDspAttrPing,
 	testDspAttrTestMissingApiKey,
 	testDspAttrTestUnknownApiKey,
 	testDspAttrTestAuthKey,
 	testDspAttrTestAuthKey2,
-	testDspAttrKillEngine,
+	testDspAttrTestAuthKey3,
 }
 
 //Test start here
 func TestDspAttributeS(t *testing.T) {
-	for _, stest := range sTestsDspAttr {
-		t.Run("", stest)
-	}
-}
-
-func testDspAttrInitCfg(t *testing.T) {
-	var err error
-	dspAttrCfgPath = path.Join(dspDataDir, "conf", "samples", "dispatchers", "dispatchers")
-	dspAttrCfg, err = config.NewCGRConfigFromFolder(dspAttrCfgPath)
-	if err != nil {
-		t.Error(err)
-	}
-	dspAttrCfg.DataFolderPath = dspDataDir // Share DataFolderPath through config towards StoreDb for Flush()
-	config.SetCgrConfig(dspAttrCfg)
-	instAttrCfgPath = path.Join(dspDataDir, "conf", "samples", "dispatchers", "attributes")
-	instAttrCfg, err = config.NewCGRConfigFromFolder(instAttrCfgPath)
-	if err != nil {
-		t.Error(err)
-	}
-	instAttrCfg.DataFolderPath = dspDataDir // Share DataFolderPath through config towards StoreDb for Flush()
-	config.SetCgrConfig(instAttrCfg)
-}
-
-func testDspAttrInitDataDb(t *testing.T) {
-	if err := engine.InitDataDb(instAttrCfg); err != nil {
-		t.Fatal(err)
-	}
-}
-
-// Wipe out the cdr database
-func testDspAttrResetStorDb(t *testing.T) {
-	if err := engine.InitStorDb(instAttrCfg); err != nil {
-		t.Fatal(err)
-	}
-}
-
-// Start CGR Engine
-func testDspAttrStartEngine(t *testing.T) {
-	if _, err := engine.StartEngine(instAttrCfgPath, dspDelay); err != nil {
-		t.Fatal(err)
-	}
-	if _, err := engine.StartEngine(dspAttrCfgPath, dspDelay); err != nil {
-		t.Fatal(err)
-	}
-}
-
-// Connect rpc client to rater
-func testDspAttrRPCConn(t *testing.T) {
-	var err error
-	instAttrRPC, err = jsonrpc.Dial("tcp", instAttrCfg.ListenCfg().RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
-	if err != nil {
-		t.Fatal(err)
-	}
-	dspAttrRPC, err = jsonrpc.Dial("tcp", dspAttrCfg.ListenCfg().RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
-	if err != nil {
-		t.Fatal(err)
-	}
-}
-
-func testDspAttrLoadData(t *testing.T) {
-	var reply string
-	attrs := &utils.AttrLoadTpFromFolder{
-		FolderPath: path.Join(dspDataDir, "tariffplans", "dispatchers")}
-	if err := instAttrRPC.Call("ApierV1.LoadTariffPlanFromFolder", attrs, &reply); err != nil {
-		t.Error(err)
-	}
+	allEngine = newTestEngine(t, path.Join(dspDataDir, "conf", "samples", "dispatchers", "all"), true, true)
+	allEngine2 = newTestEngine(t, path.Join(dspDataDir, "conf", "samples", "dispatchers", "all2"), true, true)
+	attrEngine = newTestEngine(t, path.Join(dspDataDir, "conf", "samples", "dispatchers", "attributes"), true, true)
+	dispEngine = newTestEngine(t, path.Join(dspDataDir, "conf", "samples", "dispatchers", "dispatchers"), true, true)
+	allEngine.loadData(t, path.Join(dspDataDir, "tariffplans", "tutorial"))
+	allEngine2.loadData(t, path.Join(dspDataDir, "tariffplans", "oldtutorial"))
+	attrEngine.loadData(t, path.Join(dspDataDir, "tariffplans", "dispatchers"))
 	time.Sleep(500 * time.Millisecond)
+	for _, stest := range sTestsDspAttr {
+		t.Run("TestDspAttributeS", stest)
+	}
+	attrEngine.stopEngine(t)
+	dispEngine.stopEngine(t)
+	allEngine.stopEngine(t)
+	allEngine2.stopEngine(t)
 }
 
-func testDspAttrPing(t *testing.T) {
+func testDspAttrPingFailover(t *testing.T) {
 	var reply string
-	if err := instAttrRPC.Call(utils.AttributeSv1Ping, &utils.CGREvent{}, &reply); err != nil {
+	if err := allEngine.RCP.Call(utils.AttributeSv1Ping, &utils.CGREvent{}, &reply); err != nil {
 		t.Error(err)
 	} else if reply != utils.Pong {
 		t.Errorf("Received: %s", reply)
 	}
-	if dspAttrRPC == nil {
-		t.Fatal(dspAttrRPC)
+	reply = ""
+	if err := allEngine2.RCP.Call(utils.AttributeSv1Ping, &utils.CGREvent{}, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.Pong {
+		t.Errorf("Received: %s", reply)
 	}
-	if err := dspAttrRPC.Call(utils.AttributeSv1Ping, &CGREvWithApiKey{
+	reply = ""
+	if err := dispEngine.RCP.Call(utils.AttributeSv1Ping, &CGREvWithApiKey{
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+		},
+		APIKey: "attr12345",
+	}, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.Pong {
+		t.Errorf("Received: %s", reply)
+	}
+	allEngine.stopEngine(t)
+	reply = ""
+	if err := dispEngine.RCP.Call(utils.AttributeSv1Ping, &CGREvWithApiKey{
+		CGREvent: utils.CGREvent{
+			ID:     "PING",
+			Tenant: "cgrates.org",
+		},
+		APIKey: "attr12345",
+	}, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.Pong {
+		t.Errorf("Received: %s", reply)
+	}
+	allEngine2.stopEngine(t)
+	reply = ""
+	if err := dispEngine.RCP.Call(utils.AttributeSv1Ping, &CGREvWithApiKey{
+		CGREvent: utils.CGREvent{
+			ID:     "PING",
+			Tenant: "cgrates.org",
+		},
+		APIKey: "attr12345",
+	}, &reply); err == nil {
+		t.Errorf("Expected error but recived %v and reply %v\n", err, reply)
+	}
+	allEngine.startEngine(t)
+	allEngine2.startEngine(t)
+}
+
+func testDspAttrGetAttrFailover(t *testing.T) {
+	args := &CGREvWithApiKey{
+		APIKey: "attr12345",
+		CGREvent: utils.CGREvent{
+			Tenant:  "cgrates.org",
+			ID:      "testAttributeSGetAttributeForEvent",
+			Context: utils.StringPointer("simpleauth"),
+			Event: map[string]interface{}{
+				utils.Account:    "1002",
+				utils.EVENT_NAME: "Event1",
+			},
+		},
+	}
+	eAttrPrf := &engine.AttributeProfile{
+		Tenant:    args.Tenant,
+		ID:        "ATTR_1002_SIMPLEAUTH",
+		FilterIDs: []string{"*string:Account:1002"},
+		Contexts:  []string{"simpleauth"},
+		Attributes: []*engine.Attribute{
+			{
+				FieldName:  "Password",
+				Initial:    utils.ANY,
+				Substitute: config.NewRSRParsersMustCompile("CGRateS.org", true, utils.INFIELD_SEP),
+				Append:     true,
+			},
+		},
+		Weight: 20.0,
+	}
+	eAttrPrf.Compile()
+
+	eRply := &engine.AttrSProcessEventReply{
+		MatchedProfiles: []string{"ATTR_1002_SIMPLEAUTH"},
+		AlteredFields:   []string{"Password"},
+		CGREvent: &utils.CGREvent{
+			Tenant:  "cgrates.org",
+			ID:      "testAttributeSGetAttributeForEvent",
+			Context: utils.StringPointer("simpleauth"),
+			Event: map[string]interface{}{
+				utils.Account:    "1002",
+				utils.EVENT_NAME: "Event1",
+				"Password":       "CGRateS.org",
+			},
+		},
+	}
+
+	var attrReply *engine.AttributeProfile
+	var rplyEv engine.AttrSProcessEventReply
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
+		args, &attrReply); err == nil || err.Error() != utils.ErrNotFound.Error() {
+		t.Error(err)
+	}
+
+	if err := dispEngine.RCP.Call(utils.AttributeSv1ProcessEvent,
+		args, &rplyEv); err == nil || err.Error() != utils.ErrNotFound.Error() {
+		t.Error(err)
+	} else if reflect.DeepEqual(eRply, &rplyEv) {
+		t.Errorf("Expecting: %s, received: %s",
+			utils.ToJSON(eRply), utils.ToJSON(rplyEv))
+	}
+
+	allEngine2.stopEngine(t)
+
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
+		args, &attrReply); err != nil {
+		t.Error(err)
+	}
+	if attrReply != nil {
+		attrReply.Compile()
+	}
+	if !reflect.DeepEqual(eAttrPrf, attrReply) {
+		t.Errorf("Expecting: %s, received: %s", utils.ToJSON(eAttrPrf), utils.ToJSON(attrReply))
+	}
+
+	if err := dispEngine.RCP.Call(utils.AttributeSv1ProcessEvent,
+		args, &rplyEv); err != nil {
+		t.Error(err)
+	} else if !reflect.DeepEqual(eRply, &rplyEv) {
+		t.Errorf("Expecting: %s, received: %s",
+			utils.ToJSON(eRply), utils.ToJSON(rplyEv))
+	}
+
+	allEngine2.startEngine(t)
+}
+
+func testDspAttrPing(t *testing.T) {
+	var reply string
+	if err := allEngine.RCP.Call(utils.AttributeSv1Ping, &utils.CGREvent{}, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.Pong {
+		t.Errorf("Received: %s", reply)
+	}
+	if dispEngine.RCP == nil {
+		t.Fatal(dispEngine.RCP)
+	}
+	if err := dispEngine.RCP.Call(utils.AttributeSv1Ping, &CGREvWithApiKey{
 		CGREvent: utils.CGREvent{
 			Tenant: "cgrates.org",
 		},
@@ -162,7 +312,7 @@ func testDspAttrTestMissingApiKey(t *testing.T) {
 		},
 	}
 	var attrReply *engine.AttributeProfile
-	if err := dspAttrRPC.Call(utils.AttributeSv1GetAttributeForEvent,
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
 		args, &attrReply); err == nil || err.Error() != utils.NewErrMandatoryIeMissing(utils.APIKey).Error() {
 		t.Errorf("Error:%v rply=%s", err, utils.ToJSON(attrReply))
 	}
@@ -181,7 +331,7 @@ func testDspAttrTestUnknownApiKey(t *testing.T) {
 		},
 	}
 	var attrReply *engine.AttributeProfile
-	if err := dspAttrRPC.Call(utils.AttributeSv1GetAttributeForEvent,
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
 		args, &attrReply); err == nil || err.Error() != utils.ErrUnknownApiKey.Error() {
 		t.Error(err)
 	}
@@ -200,7 +350,7 @@ func testDspAttrTestAuthKey(t *testing.T) {
 		},
 	}
 	var attrReply *engine.AttributeProfile
-	if err := dspAttrRPC.Call(utils.AttributeSv1GetAttributeForEvent,
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
 		args, &attrReply); err == nil || err.Error() != utils.ErrUnauthorizedApi.Error() {
 		t.Error(err)
 	}
@@ -235,7 +385,7 @@ func testDspAttrTestAuthKey2(t *testing.T) {
 	}
 	eAttrPrf.Compile()
 	var attrReply *engine.AttributeProfile
-	if err := dspAttrRPC.Call(utils.AttributeSv1GetAttributeForEvent,
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
 		args, &attrReply); err != nil {
 		t.Error(err)
 	}
@@ -261,7 +411,7 @@ func testDspAttrTestAuthKey2(t *testing.T) {
 	}
 
 	var rplyEv engine.AttrSProcessEventReply
-	if err := dspAttrRPC.Call(utils.AttributeSv1ProcessEvent,
+	if err := dispEngine.RCP.Call(utils.AttributeSv1ProcessEvent,
 		args, &rplyEv); err != nil {
 		t.Error(err)
 	} else if !reflect.DeepEqual(eRply, &rplyEv) {
@@ -270,11 +420,22 @@ func testDspAttrTestAuthKey2(t *testing.T) {
 	}
 }
 
-func testDspAttrKillEngine(t *testing.T) {
-	if err := engine.KillEngine(dspDelay); err != nil {
-		t.Error(err)
+func testDspAttrTestAuthKey3(t *testing.T) {
+	args := &CGREvWithApiKey{
+		APIKey: "attr12345",
+		CGREvent: utils.CGREvent{
+			Tenant:  "cgrates.org",
+			ID:      "testAttributeSGetAttributeForEvent",
+			Context: utils.StringPointer("simpleauth"),
+			Event: map[string]interface{}{
+				utils.Account:    "1001",
+				utils.EVENT_NAME: "Event1",
+			},
+		},
 	}
-	if err := engine.KillEngine(dspDelay); err != nil {
+	var attrReply *engine.AttributeProfile
+	if err := dispEngine.RCP.Call(utils.AttributeSv1GetAttributeForEvent,
+		args, &attrReply); err == nil || err.Error() != utils.ErrNotFound.Error() {
 		t.Error(err)
 	}
 }

dispatchers/chargers.go

@@ -18,40 +18,46 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 
 package dispatchers
 
-/*
 import (
 	"github.com/cgrates/cgrates/engine"
 	"github.com/cgrates/cgrates/utils"
 )
 
-func (dS *DispatcherService) ChargerSv1Ping(ign string, reply *string) error {
-	if dS.chargerS == nil {
-		return utils.NewErrNotConnected(utils.ChargerS)
+func (dS *DispatcherService) ChargerSv1Ping(args *CGREvWithApiKey, reply *string) (err error) {
+	if dS.attrS != nil {
+		if err = dS.authorize(utils.ChargerSv1Ping,
+			args.CGREvent.Tenant,
+			args.APIKey, args.CGREvent.Time); err != nil {
+			return
+		}
 	}
-	return dS.chargerS.Call(utils.ChargerSv1Ping, ign, reply)
+	return dS.Dispatch(&args.CGREvent, utils.MetaChargers,
+		utils.ChargerSv1Ping, args.CGREvent, reply)
 }
 
 func (dS *DispatcherService) ChargerSv1GetChargersForEvent(args *CGREvWithApiKey,
 	reply *engine.ChargerProfiles) (err error) {
-	if dS.chargerS == nil {
-		return utils.NewErrNotConnected(utils.ChargerS)
+	if dS.attrS != nil {
+		if err = dS.authorize(utils.ChargerSv1GetChargersForEvent,
+			args.CGREvent.Tenant,
+			args.APIKey, args.CGREvent.Time); err != nil {
+			return
+		}
 	}
-	if err = dS.authorize(utils.ChargerSv1GetChargersForEvent, args.CGREvent.Tenant,
-		args.APIKey, args.CGREvent.Time); err != nil {
-		return
-	}
-	return dS.chargerS.Call(utils.ChargerSv1GetChargersForEvent, args.CGREvent, reply)
+	return dS.Dispatch(&args.CGREvent, utils.MetaChargers,
+		utils.ChargerSv1GetChargersForEvent, args.CGREvent, reply)
 }
 
 func (dS *DispatcherService) ChargerSv1ProcessEvent(args *CGREvWithApiKey,
 	reply *[]*engine.AttrSProcessEventReply) (err error) {
-	if dS.chargerS == nil {
-		return utils.NewErrNotConnected(utils.ChargerS)
+	if dS.attrS != nil {
+		if err = dS.authorize(utils.ChargerSv1ProcessEvent,
+			args.CGREvent.Tenant,
+			args.APIKey, args.CGREvent.Time); err != nil {
+			return
+		}
+
 	}
-	if err = dS.authorize(utils.ChargerSv1ProcessEvent, args.CGREvent.Tenant,
-		args.APIKey, args.CGREvent.Time); err != nil {
-		return
-	}
-	return dS.chargerS.Call(utils.ChargerSv1ProcessEvent, args.CGREvent, reply)
+	return dS.Dispatch(&args.CGREvent, utils.MetaChargers,
+		utils.ChargerSv1ProcessEvent, args.CGREvent, reply)
 }
-*/

dispatchers/chargers_it_test.go

@@ -66,14 +66,14 @@ func TestDspChargerS(t *testing.T) {
 
 func testDspCppInitCfg(t *testing.T) {
 	var err error
-	dspCppCfgPath = path.Join(dspDataDir, "conf", "samples", "dispatcher")
+	dspCppCfgPath = path.Join(dspDataDir, "conf", "samples", "dispatchers", "dispatchers")
 	dspCppCfg, err = config.NewCGRConfigFromFolder(dspCppCfgPath)
 	if err != nil {
 		t.Error(err)
 	}
 	dspCppCfg.DataFolderPath = dspDataDir // Share DataFolderPath through config towards StoreDb for Flush()
 	config.SetCgrConfig(dspCppCfg)
-	instCppCfgPath = path.Join(dspDataDir, "conf", "samples", "tutmysql")
+	instCppCfgPath = path.Join(dspDataDir, "conf", "samples", "dispatchers", "attributes")
 	instCppCfg, err = config.NewCGRConfigFromFolder(instCppCfgPath)
 	if err != nil {
 		t.Error(err)

dispatchers/libdispatcher.go

@@ -58,6 +58,7 @@ type WeightDispatcher struct {
 func (wd *WeightDispatcher) SetProfile(pfl *engine.DispatcherProfile) {
 	pfl.Conns.Sort()
 	wd.pfl = pfl
+	wd.nextConnIdx = 0
 	return
 }
 

utils/errors.go

@@ -206,10 +206,10 @@ func IsNetworkError(err error) bool {
 			syscall.ECONNRESET.Error()) { // connection reset
 		return true
 	}
-	return err == rpc.ErrShutdown ||
-		err == ErrReqUnsynchronized ||
-		err == ErrDisconnected ||
-		err == ErrReplyTimeout ||
+	return err.Error() == rpc.ErrShutdown.Error() ||
+		err.Error() == ErrReqUnsynchronized.Error() ||
+		err.Error() == ErrDisconnected.Error() ||
+		err.Error() == ErrReplyTimeout.Error() ||
 		err.Error() == ErrSessionNotFound.Error() ||
 		strings.HasPrefix(err.Error(), "rpc: can't find service")
 }