Add support for *dynaprepaid request

apier/v1/api_interfaces_test.go

@@ -76,7 +76,7 @@ func TestGuardianSv1Interface(t *testing.T) {
 
 func TestSchedulerSv1Interface(t *testing.T) {
 	_ = SchedulerSv1Interface(NewDispatcherSchedulerSv1(nil))
-	_ = SchedulerSv1Interface(NewSchedulerSv1(nil))
+	_ = SchedulerSv1Interface(NewSchedulerSv1(nil, nil))
 }
 
 func TestCDRsV1Interface(t *testing.T) {

apier/v1/scheduler.go

@@ -20,11 +20,7 @@ package v1
 
 import (
 	"errors"
-	"fmt"
-	"sort"
-	"time"
 
-	"github.com/cgrates/cgrates/engine"
 	"github.com/cgrates/cgrates/scheduler"
 	"github.com/cgrates/cgrates/utils"
 )
@@ -112,94 +108,3 @@ func (self *APIerSv1) GetScheduledActions(args scheduler.ArgsGetScheduledActions
 	*reply = rpl
 	return nil
 }
-
-type AttrsExecuteScheduledActions struct {
-	ActionPlanID       string
-	TimeStart, TimeEnd time.Time // replay the action timings between the two dates
-}
-
-func (self *APIerSv1) ExecuteScheduledActions(attr AttrsExecuteScheduledActions, reply *string) error {
-	if attr.ActionPlanID != "" { // execute by ActionPlanID
-		apl, err := self.DataManager.GetActionPlan(attr.ActionPlanID, false, utils.NonTransactional)
-		if err != nil {
-			*reply = err.Error()
-			return err
-		}
-		if apl != nil {
-			// order by weight
-			engine.ActionTimingWeightOnlyPriorityList(apl.ActionTimings).Sort()
-			for _, at := range apl.ActionTimings {
-				if at.IsASAP() {
-					continue
-				}
-
-				at.SetAccountIDs(apl.AccountIDs) // copy the accounts
-				at.SetActionPlanID(apl.Id)
-				err := at.Execute(nil, nil)
-				if err != nil {
-					*reply = err.Error()
-					return err
-				}
-				utils.Logger.Info(fmt.Sprintf("<Force Scheduler> Executing action %s ", at.ActionsID))
-			}
-		}
-	}
-	if !attr.TimeStart.IsZero() && !attr.TimeEnd.IsZero() { // execute between two dates
-		actionPlans, err := self.DataManager.GetAllActionPlans()
-		if err != nil && err != utils.ErrNotFound {
-			err := fmt.Errorf("cannot get action plans: %v", err)
-			*reply = err.Error()
-			return err
-		}
-
-		// recreate the queue
-		queue := engine.ActionTimingPriorityList{}
-		for _, actionPlan := range actionPlans {
-			for _, at := range actionPlan.ActionTimings {
-				if at.Timing == nil {
-					continue
-				}
-				if at.IsASAP() {
-					continue
-				}
-				if at.GetNextStartTime(attr.TimeStart).Before(attr.TimeStart) {
-					// the task is obsolete, do not add it to the queue
-					continue
-				}
-				at.SetAccountIDs(actionPlan.AccountIDs) // copy the accounts
-				at.SetActionPlanID(actionPlan.Id)
-				at.ResetStartTimeCache()
-				queue = append(queue, at)
-			}
-		}
-		sort.Sort(queue)
-		// start playback execution loop
-		current := attr.TimeStart
-		for len(queue) > 0 && current.Before(attr.TimeEnd) {
-			a0 := queue[0]
-			current = a0.GetNextStartTime(current)
-			if current.Before(attr.TimeEnd) || current.Equal(attr.TimeEnd) {
-				utils.Logger.Info(fmt.Sprintf("<Replay Scheduler> Executing action %s for time %v", a0.ActionsID, current))
-				err := a0.Execute(nil, nil)
-				if err != nil {
-					*reply = err.Error()
-					return err
-				}
-				// if after execute the next start time is in the past then
-				// do not add it to the queue
-				a0.ResetStartTimeCache()
-				current = current.Add(time.Second)
-				start := a0.GetNextStartTime(current)
-				if start.Before(current) || start.After(attr.TimeEnd) {
-					queue = queue[1:]
-				} else {
-					queue = append(queue, a0)
-					queue = queue[1:]
-					sort.Sort(queue)
-				}
-			}
-		}
-	}
-	*reply = utils.OK
-	return nil
-}

apier/v1/schedulers.go

@@ -1,17 +1,14 @@
 /*
 Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
 Copyright (C) ITsysCOM GmbH
-
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.
-
 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.
-
 You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>
 */
@@ -19,18 +16,24 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package v1
 
 import (
+	"fmt"
+	"sort"
+	"time"
+
 	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
 	"github.com/cgrates/cgrates/utils"
 )
 
 // NewSchedulerSv1 retuns the API for SchedulerS
-func NewSchedulerSv1(cgrcfg *config.CGRConfig) *SchedulerSv1 {
-	return &SchedulerSv1{cgrcfg: cgrcfg}
+func NewSchedulerSv1(cgrcfg *config.CGRConfig, dm *engine.DataManager) *SchedulerSv1 {
+	return &SchedulerSv1{cgrcfg: cgrcfg, dm: dm}
 }
 
 // SchedulerSv1 is the RPC object implementing scheduler APIs
 type SchedulerSv1 struct {
 	cgrcfg *config.CGRConfig
+	dm     *engine.DataManager
 }
 
 // Reload reloads scheduler instructions
@@ -40,6 +43,132 @@ func (schdSv1 *SchedulerSv1) Reload(arg *utils.CGREventWithArgDispatcher, reply
 	return nil
 }
 
+// ExecuteActions execute an actionPlan or multiple actionsPlans between a time interval
+func (schdSv1 *SchedulerSv1) ExecuteActions(attr *utils.AttrsExecuteActions, reply *string) error {
+	if attr.ActionPlanID != "" { // execute by ActionPlanID
+		apl, err := schdSv1.dm.GetActionPlan(attr.ActionPlanID, false, utils.NonTransactional)
+		if err != nil {
+			*reply = err.Error()
+			return err
+		}
+		if apl != nil {
+			// order by weight
+			engine.ActionTimingWeightOnlyPriorityList(apl.ActionTimings).Sort()
+			for _, at := range apl.ActionTimings {
+				if at.IsASAP() {
+					continue
+				}
+
+				at.SetAccountIDs(apl.AccountIDs) // copy the accounts
+				at.SetActionPlanID(apl.Id)
+				err := at.Execute(nil, nil)
+				if err != nil {
+					*reply = err.Error()
+					return err
+				}
+				utils.Logger.Info(fmt.Sprintf("<Force Scheduler> Executing action %s ", at.ActionsID))
+			}
+		}
+	}
+	if !attr.TimeStart.IsZero() && !attr.TimeEnd.IsZero() { // execute between two dates
+		actionPlans, err := schdSv1.dm.GetAllActionPlans()
+		if err != nil && err != utils.ErrNotFound {
+			err := fmt.Errorf("cannot get action plans: %v", err)
+			*reply = err.Error()
+			return err
+		}
+
+		// recreate the queue
+		queue := engine.ActionTimingPriorityList{}
+		for _, actionPlan := range actionPlans {
+			for _, at := range actionPlan.ActionTimings {
+				if at.Timing == nil {
+					continue
+				}
+				if at.IsASAP() {
+					continue
+				}
+				if at.GetNextStartTime(attr.TimeStart).Before(attr.TimeStart) {
+					// the task is obsolete, do not add it to the queue
+					continue
+				}
+				at.SetAccountIDs(actionPlan.AccountIDs) // copy the accounts
+				at.SetActionPlanID(actionPlan.Id)
+				at.ResetStartTimeCache()
+				queue = append(queue, at)
+			}
+		}
+		sort.Sort(queue)
+		// start playback execution loop
+		current := attr.TimeStart
+		for len(queue) > 0 && current.Before(attr.TimeEnd) {
+			a0 := queue[0]
+			current = a0.GetNextStartTime(current)
+			if current.Before(attr.TimeEnd) || current.Equal(attr.TimeEnd) {
+				utils.Logger.Info(fmt.Sprintf("<Replay Scheduler> Executing action %s for time %v", a0.ActionsID, current))
+				err := a0.Execute(nil, nil)
+				if err != nil {
+					*reply = err.Error()
+					return err
+				}
+				// if after execute the next start time is in the past then
+				// do not add it to the queue
+				a0.ResetStartTimeCache()
+				current = current.Add(time.Second)
+				start := a0.GetNextStartTime(current)
+				if start.Before(current) || start.After(attr.TimeEnd) {
+					queue = queue[1:]
+				} else {
+					queue = append(queue, a0)
+					queue = queue[1:]
+					sort.Sort(queue)
+				}
+			}
+		}
+	}
+	*reply = utils.OK
+	return nil
+}
+
+// ExecuteActionPlans execute multiple actionPlans one by one
+func (schdSv1 *SchedulerSv1) ExecuteActionPlans(attr *utils.AttrsExecuteActionPlans, reply *string) (err error) {
+	// try get account
+	// if not exist set in DM
+	accID := utils.ConcatenatedKey(attr.Tenant, attr.AccountID)
+	if _, err = schdSv1.dm.GetAccount(accID); err != nil {
+		// create account if does not exist
+		account := &engine.Account{
+			ID: accID,
+		}
+		if err = schdSv1.dm.SetAccount(account); err != nil {
+			return
+		}
+	}
+	for _, apID := range attr.ActionPlanIDs {
+		apl, err := schdSv1.dm.GetActionPlan(apID, false, utils.NonTransactional)
+		if err != nil {
+			*reply = err.Error()
+			return err
+		}
+		if apl != nil {
+			// order by weight
+			engine.ActionTimingWeightOnlyPriorityList(apl.ActionTimings).Sort()
+			for _, at := range apl.ActionTimings {
+				at.SetAccountIDs(utils.NewStringMap(accID))
+				err := at.Execute(nil, nil)
+				if err != nil {
+					*reply = err.Error()
+					return err
+				}
+				utils.Logger.Info(fmt.Sprintf("<Force Scheduler> Executing action %s ", at.ActionsID))
+			}
+		}
+	}
+
+	*reply = utils.OK
+	return nil
+}
+
 // Ping returns Pong
 func (schdSv1 *SchedulerSv1) Ping(ign *utils.CGREventWithArgDispatcher, reply *string) error {
 	*reply = utils.Pong

apier/v2/cdrs_it_test.go

@@ -70,6 +70,9 @@ var (
 		testV2CDRsInitCdrDb,
 		testV2CDRsRerate,
 
+		testV2CDRsLoadTariffPlanFromFolder,
+		testv2CDRsDynaPrepaid,
+
 		testV2CDRsKillEngine,
 	}
 )
@@ -957,6 +960,47 @@ func testV2CDRsRerate(t *testing.T) {
 	}
 }
 
+func testv2CDRsDynaPrepaid(t *testing.T) {
+	var acnt engine.Account
+	if err := cdrsRpc.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "CreatedAccount"}, &acnt); err == nil ||
+		err.Error() != utils.ErrNotFound.Error() {
+		t.Error(err)
+	}
+
+	args := &engine.ArgV1ProcessEvent{
+		Flags: []string{utils.MetaRALs},
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			Event: map[string]interface{}{
+				utils.OriginID:    "testv2CDRsDynaPrepaid",
+				utils.OriginHost:  "192.168.1.1",
+				utils.Source:      "testv2CDRsDynaPrepaid",
+				utils.RequestType: utils.MetaDynaprepaid,
+				utils.Account:     "CreatedAccount",
+				utils.Subject:     "NoSubject",
+				utils.Destination: "+1234567",
+				utils.AnswerTime:  time.Date(2018, 8, 24, 16, 00, 26, 0, time.UTC),
+				utils.Usage:       time.Minute,
+			},
+		},
+	}
+
+	var reply string
+	if err := cdrsRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+
+	if err := cdrsRpc.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "CreatedAccount"}, &acnt); err != nil {
+		t.Error(err)
+	} else if acnt.BalanceMap[utils.MONETARY][0].Value != 9.9694 {
+		t.Errorf("Unexpected balance received: %+v", acnt.BalanceMap[utils.MONETARY][0])
+	}
+}
+
 func testV2CDRsKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)

config/cdrscfg.go

@@ -33,6 +33,7 @@ type CdrsCfg struct {
 	ThresholdSConns  []string
 	StatSConns       []string
 	OnlineCDRExports []string // list of CDRE templates to use for real-time CDR exports
+	SchedulerConns   []string
 }
 
 //loadFromJsonCfg loads Cdrs config from JsonCfg
@@ -114,6 +115,17 @@ func (cdrscfg *CdrsCfg) loadFromJsonCfg(jsnCdrsCfg *CdrsJsonCfg) (err error) {
 			cdrscfg.OnlineCDRExports = append(cdrscfg.OnlineCDRExports, expProfile)
 		}
 	}
+	if jsnCdrsCfg.Scheduler_conns != nil {
+		cdrscfg.SchedulerConns = make([]string, len(*jsnCdrsCfg.Scheduler_conns))
+		for idx, connID := range *jsnCdrsCfg.Scheduler_conns {
+			// if we have the connection internal we change the name so we can have internal rpc for each subsystem
+			if connID == utils.MetaInternal {
+				cdrscfg.SchedulerConns[idx] = utils.ConcatenatedKey(utils.MetaInternal, utils.MetaScheduler)
+			} else {
+				cdrscfg.SchedulerConns[idx] = connID
+			}
+		}
+	}
 
 	return nil
 }

config/config_defaults.go

@@ -257,6 +257,7 @@ const CGRATES_CFG_JSON = `
 		"*any": "*zero1ns",
 		"*voice": "*zero1s"
 	},
+	"dynaprepaid_actionplans": [],			// actionPlans to be executed in case of *dynaprepaid request type
 },
 
 
@@ -271,6 +272,7 @@ const CGRATES_CFG_JSON = `
 	"thresholds_conns": [],					// connection to ThresholdS for CDR reporting, empty to disable thresholds functionality: <""|*internal|$rpc_conns_id>
 	"stats_conns": [],						// connections to StatS for CDR reporting, empty to disable stats functionality: <""|*internal|$rpc_conns_id>
 	"online_cdr_exports":[],				// list of CDRE profiles to use for real-time CDR exports
+	"scheduler_conns": [],					// connections to SchedulerS in case of *dynaprepaid request
 },
 
 
@@ -374,6 +376,7 @@ const CGRATES_CFG_JSON = `
 		"publickey_path": "",				// the path to the public key 
 		"privatekey_path": "",				// the path to the private key
 	},
+	"scheduler_conns": [],					// connections to SchedulerS in case of *dynaprepaid request
 },
 
 

config/config_json_test.go

@@ -532,6 +532,7 @@ func TestDfRalsJsonCfg(t *testing.T) {
 			utils.ANY:   "*zero1ns",
 			utils.VOICE: "*zero1s",
 		},
+		Dynaprepaid_actionplans: &[]string{},
 	}
 	if cfg, err := dfCgrJsonCfg.RalsJsonCfg(); err != nil {
 		t.Error(err)
@@ -565,6 +566,7 @@ func TestDfCdrsJsonCfg(t *testing.T) {
 		Thresholds_conns:     &[]string{},
 		Stats_conns:          &[]string{},
 		Online_cdr_exports:   &[]string{},
+		Scheduler_conns:      &[]string{},
 	}
 	if cfg, err := dfCgrJsonCfg.CdrsJsonCfg(); err != nil {
 		t.Error(err)
@@ -701,6 +703,7 @@ func TestSmgJsonCfg(t *testing.T) {
 			Privatekey_path:     utils.StringPointer(""),
 			Publickey_path:      utils.StringPointer(""),
 		},
+		Scheduler_conns: &[]string{},
 	}
 	if cfg, err := dfCgrJsonCfg.SessionSJsonCfg(); err != nil {
 		t.Error(err)

config/config_test.go

@@ -418,37 +418,19 @@ func TestCgrCfgJSONDefaultsScheduler(t *testing.T) {
 }
 
 func TestCgrCfgJSONDefaultsCDRS(t *testing.T) {
-	emptySlice := []string{}
-	var eCdrExtr []*utils.RSRField
-	if cgrCfg.CdrsCfg().Enabled != false {
-		t.Errorf("Expecting: false , received: %+v", cgrCfg.CdrsCfg().Enabled)
+	eCdrsCfg := &CdrsCfg{
+		Enabled:         false,
+		StoreCdrs:       true,
+		SMCostRetries:   5,
+		ChargerSConns:   []string{},
+		RaterConns:      []string{},
+		AttributeSConns: []string{},
+		ThresholdSConns: []string{},
+		StatSConns:      []string{},
+		SchedulerConns:  []string{},
 	}
-	if !reflect.DeepEqual(eCdrExtr, cgrCfg.CdrsCfg().ExtraFields) {
-		t.Errorf("Expecting: %+v , received: %+v", eCdrExtr, cgrCfg.CdrsCfg().ExtraFields)
-	}
-	if cgrCfg.CdrsCfg().StoreCdrs != true {
-		t.Errorf("Expecting: true , received: %+v", cgrCfg.CdrsCfg().StoreCdrs)
-	}
-	if cgrCfg.CdrsCfg().SMCostRetries != 5 {
-		t.Errorf("Expecting: 5 , received: %+v", cgrCfg.CdrsCfg().SMCostRetries)
-	}
-	if !reflect.DeepEqual(cgrCfg.CdrsCfg().RaterConns, emptySlice) {
-		t.Errorf("Expecting: %+v , received: %+v", emptySlice, cgrCfg.CdrsCfg().RaterConns)
-	}
-	if !reflect.DeepEqual(cgrCfg.CdrsCfg().ChargerSConns, emptySlice) {
-		t.Errorf("Expecting: %+v , received: %+v", emptySlice, cgrCfg.CdrsCfg().ChargerSConns)
-	}
-	if !reflect.DeepEqual(cgrCfg.CdrsCfg().AttributeSConns, emptySlice) {
-		t.Errorf("Expecting: %+v , received: %+v", emptySlice, cgrCfg.CdrsCfg().AttributeSConns)
-	}
-	if !reflect.DeepEqual(cgrCfg.CdrsCfg().ThresholdSConns, emptySlice) {
-		t.Errorf("Expecting: %+v , received: %+v", emptySlice, cgrCfg.CdrsCfg().ThresholdSConns)
-	}
-	if !reflect.DeepEqual(cgrCfg.CdrsCfg().StatSConns, emptySlice) {
-		t.Errorf("Expecting: %+v , received: %+v", emptySlice, cgrCfg.CdrsCfg().StatSConns)
-	}
-	if cgrCfg.CdrsCfg().OnlineCDRExports != nil {
-		t.Errorf("Expecting: nil , received: %+v", cgrCfg.CdrsCfg().OnlineCDRExports)
+	if !reflect.DeepEqual(eCdrsCfg, cgrCfg.cdrsCfg) {
+		t.Errorf("Expecting: %+v , received: %+v", eCdrsCfg, cgrCfg.cdrsCfg)
 	}
 }
 
@@ -626,6 +608,7 @@ func TestCgrCfgJSONDefaultsSMGenericCfg(t *testing.T) {
 			PayloadMaxduration: -1,
 			DefaultAttest:      "A",
 		},
+		SchedulerConns: []string{},
 	}
 	if !reflect.DeepEqual(eSessionSCfg, cgrCfg.sessionSCfg) {
 		t.Errorf("expecting: %s, received: %s",

config/libconfig_json.go

@@ -128,6 +128,7 @@ type RalsJsonCfg struct {
 	Max_computed_usage         *map[string]string
 	Max_increments             *int
 	Balance_rating_subject     *map[string]string
+	Dynaprepaid_actionplans    *[]string
 }
 
 // Scheduler config section
@@ -149,6 +150,7 @@ type CdrsJsonCfg struct {
 	Thresholds_conns     *[]string
 	Stats_conns          *[]string
 	Online_cdr_exports   *[]string
+	Scheduler_conns      *[]string
 }
 
 // Cdre config section
@@ -220,6 +222,7 @@ type SessionSJsonCfg struct {
 	Terminate_attempts    *int
 	Alterable_fields      *[]string
 	Min_dur_low_balance   *string
+	Scheduler_conns       *[]string
 	Stir                  *STIRJsonCfg
 }
 

config/ralscfg.go

@@ -35,6 +35,7 @@ type RalsCfg struct {
 	MaxComputedUsage        map[string]time.Duration
 	BalanceRatingSubject    map[string]string
 	MaxIncrements           int
+	DynaprepaidActionPlans  []string
 }
 
 //loadFromJsonCfg loads Rals config from JsonCfg
@@ -99,6 +100,12 @@ func (ralsCfg *RalsCfg) loadFromJsonCfg(jsnRALsCfg *RalsJsonCfg) (err error) {
 			ralsCfg.BalanceRatingSubject[k] = v
 		}
 	}
+	if jsnRALsCfg.Dynaprepaid_actionplans != nil {
+		ralsCfg.DynaprepaidActionPlans = make([]string, len(*jsnRALsCfg.Dynaprepaid_actionplans))
+		for i, val := range *jsnRALsCfg.Dynaprepaid_actionplans {
+			ralsCfg.DynaprepaidActionPlans[i] = val
+		}
+	}
 
 	return nil
 }

config/sessionscfg.go

@@ -99,6 +99,7 @@ type SessionSCfg struct {
 	TerminateAttempts   int
 	AlterableFields     *utils.StringSet
 	MinDurLowBalance    time.Duration
+	SchedulerConns      []string
 	STIRCfg             *STIRcfg
 }
 
@@ -276,6 +277,17 @@ func (scfg *SessionSCfg) loadFromJsonCfg(jsnCfg *SessionSJsonCfg) (err error) {
 			return err
 		}
 	}
+	if jsnCfg.Scheduler_conns != nil {
+		scfg.SchedulerConns = make([]string, len(*jsnCfg.Scheduler_conns))
+		for idx, connID := range *jsnCfg.Scheduler_conns {
+			// if we have the connection internal we change the name so we can have internal rpc for each subsystem
+			if connID == utils.MetaInternal {
+				scfg.SchedulerConns[idx] = utils.ConcatenatedKey(utils.MetaInternal, utils.MetaScheduler)
+			} else {
+				scfg.SchedulerConns[idx] = connID
+			}
+		}
+	}
 	return scfg.STIRCfg.loadFromJSONCfg(jsnCfg.Stir)
 }
 

console/scheduler_execute.go

@@ -19,15 +19,14 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package console
 
 import (
-	"github.com/cgrates/cgrates/apier/v1"
 	"github.com/cgrates/cgrates/utils"
 )
 
 func init() {
 	c := &CmdExecuteScheduledActions{
 		name:      "scheduler_execute",
-		rpcMethod: utils.APIerSv1ExecuteScheduledActions,
-		rpcParams: &v1.AttrsExecuteScheduledActions{},
+		rpcMethod: utils.SchedulerSv1ExecuteActions,
+		rpcParams: &utils.AttrsExecuteActions{},
 	}
 	commands[c.Name()] = c
 	c.CommandExecuter = &CommandExecuter{c}
@@ -37,7 +36,7 @@ func init() {
 type CmdExecuteScheduledActions struct {
 	name      string
 	rpcMethod string
-	rpcParams *v1.AttrsExecuteScheduledActions
+	rpcParams *utils.AttrsExecuteActions
 	*CommandExecuter
 }
 
@@ -51,7 +50,7 @@ func (self *CmdExecuteScheduledActions) RpcMethod() string {
 
 func (self *CmdExecuteScheduledActions) RpcParams(reset bool) interface{} {
 	if reset || self.rpcParams == nil {
-		self.rpcParams = &v1.AttrsExecuteScheduledActions{}
+		self.rpcParams = &utils.AttrsExecuteActions{}
 	}
 	return self.rpcParams
 }

data/conf/samples/cdrsv2internal/cgrates.json

@@ -9,13 +9,14 @@
 },
 
 "stor_db": {
-	"db_type": "*internal",					// stor database type to use: <mysql|postgres>
+	"db_type": "*internal",
 },
 
 
 "rals": {
-	"enabled": true,						// enable Rater service: <true|false>
+	"enabled": true,
 	"thresholds_conns": ["*localhost"],
+	"dynaprepaid_actionplans": ["PACKAGE_1001"],
 },
 
 "schedulers": {
@@ -29,6 +30,7 @@
 	"rals_conns": ["*localhost"],
 	"stats_conns": ["*localhost"],
 	"thresholds_conns": ["*localhost"],
+	"scheduler_conns": ["*localhost"],
 },
 
 "attributes": {

data/conf/samples/cdrsv2mongo/cgrates.json

@@ -9,14 +9,15 @@
 },
 
 "stor_db": {
-	"db_type": "mongo",					// stor database type to use: <mysql|postgres>
-	"db_port": 27017, 						// the port to reach the stordb
+	"db_type": "mongo",
+	"db_port": 27017,
 },
 
 
 "rals": {
-	"enabled": true,						// enable Rater service: <true|false>
+	"enabled": true,
 	"thresholds_conns": ["*localhost"],
+	"dynaprepaid_actionplans": ["PACKAGE_1001"],
 },
 
 "schedulers": {
@@ -30,6 +31,7 @@
 	"rals_conns": ["*localhost"],
 	"stats_conns": ["*localhost"],
 	"thresholds_conns": ["*localhost"],
+	"scheduler_conns": ["*localhost"],
 },
 
 "attributes": {

data/conf/samples/cdrsv2mongo_gob/cgrates.json

@@ -18,14 +18,15 @@
 
 
 "stor_db": {
-	"db_type": "mongo",					// stor database type to use: <mysql|postgres>
-	"db_port": 27017, 						// the port to reach the stordb
+	"db_type": "mongo",
+	"db_port": 27017,
 },
 
 
 "rals": {
-	"enabled": true,						// enable Rater service: <true|false>
+	"enabled": true,
 	"thresholds_conns": ["conn1"],
+	"dynaprepaid_actionplans": ["PACKAGE_1001"],
 },
 
 "schedulers": {
@@ -39,6 +40,7 @@
 	"rals_conns": ["conn1"],
 	"stats_conns": ["conn1"],
 	"thresholds_conns": ["conn1"],
+	"scheduler_conns": ["conn1"],
 },
 
 "attributes": {

data/conf/samples/cdrsv2mysql/cdrsv2mysql.json

@@ -17,6 +17,7 @@
 "rals": {
 	"enabled": true,
 	"thresholds_conns": ["*localhost"],
+	"dynaprepaid_actionplans": ["PACKAGE_1001"],
 },
 
 "schedulers": {
@@ -30,6 +31,7 @@
 	"rals_conns": ["*localhost"],
 	"stats_conns": ["*localhost"],
 	"thresholds_conns": ["*localhost"],
+	"scheduler_conns": ["*localhost"],
 },
 
 "attributes": {

data/conf/samples/cdrsv2mysql_gob/cdrsv2mysql.json

@@ -25,6 +25,7 @@
 "rals": {
 	"enabled": true,
 	"thresholds_conns": ["conn1"],
+	"dynaprepaid_actionplans": ["PACKAGE_1001"],
 },
 
 "schedulers": {
@@ -38,6 +39,7 @@
 	"rals_conns": ["conn1"],
 	"stats_conns": ["conn1"],
 	"thresholds_conns": ["conn1"],
+	"scheduler_conns": ["*localhost"],
 },
 
 "attributes": {

data/conf/samples/cdrsv2psql/cdrsv2psql.json

@@ -9,14 +9,15 @@
 },
 
 "stor_db": {
-	"db_type": "postgres",					// stor database type to use: <mysql|postgres>
-	"db_port": 5432, 						// the port to reach the stordb
+	"db_type": "postgres",
+	"db_port": 5432,
 	"db_password": "CGRateS.org"
 },
 
 "rals": {
-	"enabled": true,						// enable Rater service: <true|false>
+	"enabled": true,
 	"thresholds_conns": ["*localhost"],
+	"dynaprepaid_actionplans": ["PACKAGE_1001"],
 },
 
 "schedulers": {
@@ -30,6 +31,7 @@
 	"rals_conns": ["*localhost"],
 	"stats_conns": ["*localhost"],
 	"thresholds_conns": ["*localhost"],
+	"scheduler_conns": ["*localhost"],
 },
 
 "attributes": {

engine/cdrs.go

@@ -229,7 +229,7 @@ func (cdrS *CDRServer) rateCDR(cdr *CDRWithArgDispatcher) ([]*CDR, error) {
 }
 
 var reqTypes = utils.NewStringSet([]string{utils.META_PSEUDOPREPAID, utils.META_POSTPAID, utils.META_PREPAID,
-	utils.PSEUDOPREPAID, utils.POSTPAID, utils.PREPAID})
+	utils.PSEUDOPREPAID, utils.POSTPAID, utils.PREPAID, utils.MetaDynaprepaid})
 
 // getCostFromRater will retrieve the cost from RALs
 func (cdrS *CDRServer) getCostFromRater(cdr *CDRWithArgDispatcher) (*CallCost, error) {
@@ -259,6 +259,23 @@ func (cdrS *CDRServer) getCostFromRater(cdr *CDRWithArgDispatcher) (*CallCost, e
 			utils.ResponderDebit,
 			&CallDescriptorWithArgDispatcher{CallDescriptor: cd,
 				ArgDispatcher: cdr.ArgDispatcher}, cc)
+		if err != nil && err.Error() == utils.ErrAccountNotFound.Error() &&
+			cdr.RequestType == utils.MetaDynaprepaid {
+			var reply string
+			// execute the actionPlan configured in RalS
+			if err = cdrS.connMgr.Call(cdrS.cgrCfg.CdrsCfg().SchedulerConns, nil,
+				utils.SchedulerSv1ExecuteActionPlans, &utils.AttrsExecuteActionPlans{
+					ActionPlanIDs: cdrS.cgrCfg.RalsCfg().DynaprepaidActionPlans,
+					AccountID:     cdr.Account, Tenant: cdr.Tenant},
+				&reply); err != nil {
+				return cc, err
+			}
+			// execute again the Debit operation
+			err = cdrS.connMgr.Call(cdrS.cgrCfg.CdrsCfg().RaterConns, nil,
+				utils.ResponderDebit,
+				&CallDescriptorWithArgDispatcher{CallDescriptor: cd,
+					ArgDispatcher: cdr.ArgDispatcher}, cc)
+		}
 	} else {
 		err = cdrS.connMgr.Call(cdrS.cgrCfg.CdrsCfg().RaterConns, nil,
 			utils.ResponderGetCost,

services/schedulers.go

@@ -81,7 +81,7 @@ func (schS *SchedulerService) Start() (err error) {
 	schS.schS = scheduler.NewScheduler(datadb, schS.cfg, fltrS)
 	go schS.schS.Loop()
 
-	schS.rpc = v1.NewSchedulerSv1(schS.cfg)
+	schS.rpc = v1.NewSchedulerSv1(schS.cfg, datadb)
 	if !schS.cfg.DispatcherSCfg().Enabled {
 		schS.server.RpcRegister(schS.rpc)
 	}

sessions/sessions.go

@@ -409,13 +409,35 @@ func (sS *SessionS) debitSession(s *Session, sRunIdx int, dur time.Duration,
 	cd := sr.CD.Clone()
 	argDsp := s.ArgDispatcher
 	cc := new(engine.CallCost)
-	if err := sS.connMgr.Call(sS.cgrCfg.SessionSCfg().RALsConns, nil,
+	err = sS.connMgr.Call(sS.cgrCfg.SessionSCfg().RALsConns, nil,
 		utils.ResponderMaxDebit,
 		&engine.CallDescriptorWithArgDispatcher{
 			CallDescriptor: cd,
-			ArgDispatcher:  argDsp}, cc); err != nil {
-		sr.ExtraDuration += dbtRsrv
-		return 0, err
+			ArgDispatcher:  argDsp}, cc)
+	if err != nil {
+		// verify in case of *dynaprepaid RequestType
+		if err.Error() == utils.ErrAccountNotFound.Error() &&
+			sr.Event.GetStringIgnoreErrors(utils.RequestType) == utils.MetaDynaprepaid {
+			var reply string
+			// execute the actionPlan configured in RalS
+			if err = sS.connMgr.Call(sS.cgrCfg.SessionSCfg().SchedulerConns, nil,
+				utils.SchedulerSv1ExecuteActionPlans, &utils.AttrsExecuteActionPlans{
+					ActionPlanIDs: sS.cgrCfg.RalsCfg().DynaprepaidActionPlans,
+					Tenant:        cd.Tenant, AccountID: cd.Account},
+				&reply); err != nil {
+				return
+			}
+			// execute again the MaxDebit operation
+			err = sS.connMgr.Call(sS.cgrCfg.SessionSCfg().RALsConns, nil,
+				utils.ResponderMaxDebit,
+				&engine.CallDescriptorWithArgDispatcher{
+					CallDescriptor: cd,
+					ArgDispatcher:  argDsp}, cc)
+		}
+		if err != nil {
+			sr.ExtraDuration += dbtRsrv
+			return 0, err
+		}
 	}
 	sr.CD.TimeEnd = cc.GetEndTime() // set debited timeEnd
 	ccDuration := cc.GetDuration()

utils/apitpdata.go

@@ -1442,3 +1442,13 @@ type ArgCacheReplicateRemove struct {
 	*ArgDispatcher
 	TenantArg
 }
+
+type AttrsExecuteActions struct {
+	ActionPlanID       string
+	TimeStart, TimeEnd time.Time // replay the action timings between the two dates
+}
+
+type AttrsExecuteActionPlans struct {
+	ActionPlanIDs     []string
+	Tenant, AccountID string
+}

utils/consts.go

@@ -717,6 +717,7 @@ const (
 	MetaPAP                   = "*pap"
 	MetaCHAP                  = "*chap"
 	MetaMSCHAPV2              = "*mschapv2"
+	MetaDynaprepaid           = "*dynaprepaid"
 )
 
 // Migrator Action
@@ -1459,9 +1460,11 @@ const (
 
 // Scheduler
 const (
-	SchedulerSv1       = "SchedulerSv1"
-	SchedulerSv1Ping   = "SchedulerSv1.Ping"
-	SchedulerSv1Reload = "SchedulerSv1.Reload"
+	SchedulerSv1                   = "SchedulerSv1"
+	SchedulerSv1Ping               = "SchedulerSv1.Ping"
+	SchedulerSv1Reload             = "SchedulerSv1.Reload"
+	SchedulerSv1ExecuteActions     = "SchedulerSv1.ExecuteActions"
+	SchedulerSv1ExecuteActionPlans = "SchedulerSv1.ExecuteActionPlans"
 )
 
 //cgr_ variables