Added tests for dispatcher connection as default internal connection

apier/v1/dispatcher.go

@@ -297,22 +297,22 @@ func (dRs *DispatcherResourceSv1) Ping(args *utils.CGREventWithArgDispatcher, re
 }
 
 // GetResourcesForEvent implements ResourceSv1GetResourcesForEvent
-func (dRs *DispatcherResourceSv1) GetResourcesForEvent(args *utils.ArgRSv1ResourceUsage,
+func (dRs *DispatcherResourceSv1) GetResourcesForEvent(args utils.ArgRSv1ResourceUsage,
 	reply *engine.Resources) error {
 	return dRs.dRs.ResourceSv1GetResourcesForEvent(args, reply)
 }
 
-func (dRs *DispatcherResourceSv1) AuthorizeResources(args *utils.ArgRSv1ResourceUsage,
+func (dRs *DispatcherResourceSv1) AuthorizeResources(args utils.ArgRSv1ResourceUsage,
 	reply *string) error {
 	return dRs.dRs.ResourceSv1AuthorizeResources(args, reply)
 }
 
-func (dRs *DispatcherResourceSv1) AllocateResources(args *utils.ArgRSv1ResourceUsage,
+func (dRs *DispatcherResourceSv1) AllocateResources(args utils.ArgRSv1ResourceUsage,
 	reply *string) error {
 	return dRs.dRs.ResourceSv1AllocateResources(args, reply)
 }
 
-func (dRs *DispatcherResourceSv1) ReleaseResources(args *utils.ArgRSv1ResourceUsage,
+func (dRs *DispatcherResourceSv1) ReleaseResources(args utils.ArgRSv1ResourceUsage,
 	reply *string) error {
 	return dRs.dRs.ResourceSv1ReleaseResources(args, reply)
 }

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

@@ -40,6 +40,11 @@
 	"enabled": true,
 },
 
+"sessions": {
+	"enabled": true,
+	"listen_bijson": ":3014",
+},
+
 "dispatchers":{
  	"enabled": true,
  	"attributes_conns": [

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

@@ -48,6 +48,11 @@
 	"enabled": true,
 },
 
+"sessions": {
+	"enabled": true,
+	"listen_bijson": ":3014",
+},
+
 "dispatchers":{
  	"enabled": true,
  	"attributes_conns": [

data/tariffplans/dispatchers/Attributes.csv

@@ -1,5 +1,6 @@
 #Tenant,ID,Contexts,FilterIDs,ActivationInterval,AttributeFilterIDs,FieldName,Type,Value,Blocker,Weight
 cgrates.org,ATTR_1001_SIMPLEAUTH,*any,*string:~Account:1001,,,Password,*constant,CGRateS.org,false,20
+cgrates.org,ATTR_1001_SIMPLEAUTH,*any,,,,EventName,*constant,*none,false,20
 cgrates.org,ATTR_1003_SIMPLEAUTH,*any,*string:~Account:1003,,,Password,*constant,CGRateS.com,false,20
 cgrates.org,ATTR_API_ATTR_FAKE_AUTH,*auth,*string:~APIKey:12345,,,APIMethods,*constant,,false,20
 cgrates.org,ATTR_API_ATTR_AUTH,*auth,*string:~APIKey:attr12345,,,APIMethods,*constant,AttributeSv1.Ping&AttributeSv1.GetAttributeForEvent&AttributeSv1.ProcessEvent,false,20
@@ -15,3 +16,4 @@ cgrates.org,ATTR_API_GRD_AUTH,*auth,*string:~APIKey:grd12345,,,APIMethods,*const
 cgrates.org,ATTR_API_SCHD_AUTH,*auth,*string:~APIKey:sched12345,,,APIMethods,*constant,SchedulerSv1.Ping,false,20
 cgrates.org,ATTR_API_CDRS_AUTH,*auth,*string:~APIKey:cdrs12345,,,APIMethods,*constant,CDRsV1.Ping&CDRsV1.ProcessEvent&CDRsV1.GetCDRs&CDRsV1.CountCDRs&CDRsV1.ProcessCDR&CDRsV1.ProcessExternalCDR,false,20
 cgrates.org,ATTR_API_DSP_AUTH,*auth,*string:~APIKey:dsp12345,,,APIMethods,*constant,DispatcherSv1.Ping&DispatcherSv1.GetProfileForEvent,false,20
+cgrates.org,ATTR_API_PSE_AUTH,*auth,*string:~APIKey:pse12345,,,APIMethods,*constant,SessionSv1.Ping&SessionSv1.AuthorizeEvent&SessionSv1.AuthorizeEventWithDigest&SessionSv1.InitiateSession&SessionSv1.InitiateSessionWithDigest&SessionSv1.UpdateSession&SessionSv1.SyncSessions&SessionSv1.TerminateSession&SessionSv1.ProcessCDR&SessionSv1.ProcessEvent&SessionSv1.GetActiveSessions&SessionSv1.GetActiveSessionsCount&SessionSv1.ForceDisconnect&SessionSv1.GetPassiveSessions&SessionSv1.GetPassiveSessionsCount&SessionSv1.ReplicateSessions&SessionSv1.SetPassiveSession&AttributeSv1.ProcessEvent&Responder.Debit&ResourceSv1.AllocateResources&ChargerSv1.ProcessEvent&Responder.MaxDebit,false,20

dispatchers/resources.go

@@ -38,7 +38,7 @@ func (dS *DispatcherService) ResourceSv1Ping(args *utils.CGREventWithArgDispatch
 		utils.ResourceSv1Ping, args.CGREvent, rpl)
 }
 
-func (dS *DispatcherService) ResourceSv1GetResourcesForEvent(args *utils.ArgRSv1ResourceUsage,
+func (dS *DispatcherService) ResourceSv1GetResourcesForEvent(args utils.ArgRSv1ResourceUsage,
 	reply *engine.Resources) (err error) {
 	if args.ArgDispatcher == nil {
 		return utils.NewErrMandatoryIeMissing("ArgDispatcher")
@@ -55,7 +55,7 @@ func (dS *DispatcherService) ResourceSv1GetResourcesForEvent(args *utils.ArgRSv1
 		utils.ResourceSv1GetResourcesForEvent, args, reply)
 }
 
-func (dS *DispatcherService) ResourceSv1AuthorizeResources(args *utils.ArgRSv1ResourceUsage,
+func (dS *DispatcherService) ResourceSv1AuthorizeResources(args utils.ArgRSv1ResourceUsage,
 	reply *string) (err error) {
 	if args.ArgDispatcher == nil {
 		return utils.NewErrMandatoryIeMissing("ArgDispatcher")
@@ -72,7 +72,7 @@ func (dS *DispatcherService) ResourceSv1AuthorizeResources(args *utils.ArgRSv1Re
 		utils.ResourceSv1AuthorizeResources, args, reply)
 }
 
-func (dS *DispatcherService) ResourceSv1AllocateResources(args *utils.ArgRSv1ResourceUsage,
+func (dS *DispatcherService) ResourceSv1AllocateResources(args utils.ArgRSv1ResourceUsage,
 	reply *string) (err error) {
 	if args.ArgDispatcher == nil {
 		return utils.NewErrMandatoryIeMissing("ArgDispatcher")
@@ -89,7 +89,7 @@ func (dS *DispatcherService) ResourceSv1AllocateResources(args *utils.ArgRSv1Res
 		utils.ResourceSv1AllocateResources, args, reply)
 }
 
-func (dS *DispatcherService) ResourceSv1ReleaseResources(args *utils.ArgRSv1ResourceUsage,
+func (dS *DispatcherService) ResourceSv1ReleaseResources(args utils.ArgRSv1ResourceUsage,
 	reply *string) (err error) {
 	if args.ArgDispatcher == nil {
 		return utils.NewErrMandatoryIeMissing("ArgDispatcher")

dispatchers/sessions_it_test.go

@@ -48,6 +48,7 @@ var sTestsDspSession = []func(t *testing.T){
 	testDspSessionTerminate,
 	testDspSessionProcessCDR,
 	testDspSessionProcessEvent,
+	testDspSessionProcessEvent2,
 
 	testDspSessionReplicate,
 	testDspSessionPassive,
@@ -584,6 +585,73 @@ func testDspSessionProcessEvent(t *testing.T) {
 	}
 }
 
+func testDspSessionProcessEvent2(t *testing.T) {
+	initUsage := 5 * time.Minute
+	args := sessions.V1ProcessEventArgs{
+		AllocateResources: true,
+		Debit:             true,
+		GetAttributes:     true,
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			ID:     "TestSSv1ItProcessEvent",
+			Event: map[string]interface{}{
+				utils.CGRID:       "c87609aa1cb6e9529ab1836cfeeebaab7aa7ebaf",
+				utils.Tenant:      "cgrates.org",
+				utils.Category:    "call",
+				utils.ToR:         utils.VOICE,
+				utils.OriginID:    "TestSSv1It2",
+				utils.RequestType: utils.META_PREPAID,
+				utils.Account:     "1001",
+				utils.Destination: "1002",
+				utils.SetupTime:   time.Date(2018, time.January, 7, 16, 60, 0, 0, time.UTC),
+				utils.AnswerTime:  time.Date(2018, time.January, 7, 16, 60, 10, 0, time.UTC),
+				utils.Usage:       initUsage,
+				utils.EVENT_NAME:  "Internal",
+			},
+		},
+		ArgDispatcher: &utils.ArgDispatcher{
+			APIKey: utils.StringPointer("pse12345"),
+		},
+	}
+	var rply sessions.V1ProcessEventReply
+	if err := dispEngine.RCP.Call(utils.SessionSv1ProcessEvent,
+		args, &rply); err != nil {
+		t.Fatal(err)
+	}
+	if *rply.MaxUsage != initUsage {
+		t.Errorf("Unexpected MaxUsage: %v", rply.MaxUsage)
+	}
+	if *rply.ResourceAllocation != "RES_ACNT_1001" {
+		t.Errorf("Unexpected ResourceAllocation: %s", *rply.ResourceAllocation)
+	}
+	eAttrs := &engine.AttrSProcessEventReply{
+		MatchedProfiles: []string{"ATTR_1001_SIMPLEAUTH"},
+		AlteredFields:   []string{"Password", "EventName"},
+		CGREvent: &utils.CGREvent{
+			Tenant: "cgrates.org",
+			ID:     "TestSSv1ItProcessEvent",
+			Event: map[string]interface{}{
+				utils.CGRID:       "c87609aa1cb6e9529ab1836cfeeebaab7aa7ebaf",
+				utils.Tenant:      "cgrates.org",
+				utils.Category:    "call",
+				utils.ToR:         utils.VOICE,
+				utils.Account:     "1001",
+				utils.Destination: "1002",
+				"Password":        "CGRateS.org",
+				utils.OriginID:    "TestSSv1It2",
+				utils.RequestType: utils.META_PREPAID,
+				utils.SetupTime:   "2018-01-07T17:00:00Z",
+				utils.AnswerTime:  "2018-01-07T17:00:10Z",
+				utils.Usage:       300000000000.0,
+			},
+		},
+	}
+	if !reflect.DeepEqual(eAttrs, rply.Attributes) {
+		t.Errorf("expecting: %+v, received: %+v",
+			utils.ToJSON(eAttrs), utils.ToJSON(rply.Attributes))
+	}
+}
+
 func testDspSessionReplicate(t *testing.T) {
 	allEngine.initDataDb(t)
 	allEngine.resetStorDb(t)

sessions/sessions.go

@@ -1222,7 +1222,7 @@ func (sS *SessionS) authSession(tnt string, evStart *engine.SafEvent) (maxUsage
 
 // initSession handles a new session
 func (sS *SessionS) initSession(tnt string, evStart *engine.SafEvent, clntConnID string,
-	resID string, dbtItval time.Duration) (s *Session, err error) {
+	resID string, dbtItval time.Duration, argDisp *utils.ArgDispatcher) (s *Session, err error) {
 	cgrID := GetSetCGRID(evStart)
 	s = &Session{
 		CGRID:         cgrID,
@@ -1231,24 +1231,7 @@ func (sS *SessionS) initSession(tnt string, evStart *engine.SafEvent, clntConnID
 		EventStart:    evStart,
 		ClientConnID:  clntConnID,
 		DebitInterval: dbtItval,
-	}
-	//check if we have APIKey in event and in case it has add it in ArgDispatcher
-	apiKeyIface, errApiKey := evStart.FieldAsString([]string{utils.MetaApiKey})
-	if errApiKey == nil {
-		s.ArgDispatcher = &utils.ArgDispatcher{
-			APIKey: utils.StringPointer(apiKeyIface),
-		}
-	}
-	//check if we have RouteID in event and in case it has add it in ArgDispatcher
-	routeIDIface, errRouteID := evStart.FieldAsString([]string{utils.MetaRouteID})
-	if errRouteID == nil {
-		if errApiKey.Error() == utils.ErrNotFound.Error() { //in case we don't have APIKey, but we have RouteID we need to initialize the struct
-			s.ArgDispatcher = &utils.ArgDispatcher{
-				RouteID: utils.StringPointer(routeIDIface),
-			}
-		} else {
-			s.ArgDispatcher.RouteID = utils.StringPointer(routeIDIface)
-		}
+		ArgDispatcher: argDisp,
 	}
 	if err = sS.forkSession(s); err != nil {
 		return nil, err
@@ -1374,10 +1357,10 @@ func (sS *SessionS) endSession(s *Session, tUsage, lastUsage *time.Duration) (er
 }
 
 // chargeEvent will charge a single event (ie: SMS)
-func (sS *SessionS) chargeEvent(tnt string, ev *engine.SafEvent) (maxUsage time.Duration, err error) {
+func (sS *SessionS) chargeEvent(tnt string, ev *engine.SafEvent, argDisp *utils.ArgDispatcher) (maxUsage time.Duration, err error) {
 	cgrID := GetSetCGRID(ev)
 	var s *Session
-	if s, err = sS.initSession(tnt, ev, "", "", 0); err != nil {
+	if s, err = sS.initSession(tnt, ev, "", "", 0, argDisp); err != nil {
 		return
 	}
 	if maxUsage, err = sS.updateSession(s, ev.AsMapInterface()); err != nil {
@@ -1982,7 +1965,7 @@ func (sS *SessionS) BiRPCv1InitiateSession(clnt rpcclient.RpcClientConnection,
 			}
 		}
 		s, err := sS.initSession(args.CGREvent.Tenant, ev,
-			sS.biJClntID(clnt), originID, dbtItvl)
+			sS.biJClntID(clnt), originID, dbtItvl, args.ArgDispatcher)
 		if err != nil {
 			return utils.NewErrRALs(err)
 		}
@@ -2198,7 +2181,7 @@ func (sS *SessionS) BiRPCv1UpdateSession(clnt rpcclient.RpcClientConnection,
 		if len(ss) == 0 {
 			if s, err = sS.initSession(args.CGREvent.Tenant,
 				ev, sS.biJClntID(clnt),
-				me.GetStringIgnoreErrors(utils.OriginID), dbtItvl); err != nil {
+				me.GetStringIgnoreErrors(utils.OriginID), dbtItvl, args.ArgDispatcher); err != nil {
 				return utils.NewErrRALs(err)
 			}
 		} else {
@@ -2290,7 +2273,7 @@ func (sS *SessionS) BiRPCv1TerminateSession(clnt rpcclient.RpcClientConnection,
 		if len(ss) == 0 {
 			if s, err = sS.initSession(args.CGREvent.Tenant,
 				ev, sS.biJClntID(clnt),
-				me.GetStringIgnoreErrors(utils.OriginID), dbtItvl); err != nil {
+				me.GetStringIgnoreErrors(utils.OriginID), dbtItvl, args.ArgDispatcher); err != nil {
 				return utils.NewErrRALs(err)
 			}
 		} else {
@@ -2584,7 +2567,7 @@ func (sS *SessionS) BiRPCv1ProcessEvent(clnt rpcclient.RpcClientConnection,
 	}
 	if args.Debit {
 		if maxUsage, err := sS.chargeEvent(args.CGREvent.Tenant,
-			engine.NewSafEvent(args.CGREvent.Event)); err != nil {
+			engine.NewSafEvent(args.CGREvent.Event), args.ArgDispatcher); err != nil {
 			return utils.NewErrRALs(err)
 		} else {
 			rply.MaxUsage = &maxUsage