Add Scheduler API's in Dispatcher and add test for Scheduler ExecuteActions

apier/v1/api_interfaces.go

@@ -136,6 +136,8 @@ type GuardianSv1Interface interface {
 type SchedulerSv1Interface interface {
 	Reload(arg *utils.CGREventWithArgDispatcher, reply *string) error
 	Ping(ign *utils.CGREventWithArgDispatcher, reply *string) error
+	ExecuteActions(attr *utils.AttrsExecuteActions, reply *string) error
+	ExecuteActionPlans(attr *utils.AttrsExecuteActionPlans, reply *string) error
 }
 
 type CDRsV1Interface interface {

apier/v1/dispatcher.go

@@ -703,6 +703,16 @@ func (dS *DispatcherSchedulerSv1) Ping(args *utils.CGREventWithArgDispatcher, re
 	return dS.dS.SchedulerSv1Ping(args, reply)
 }
 
+// ExecuteActions execute an actionPlan or multiple actionsPlans between a time interval
+func (dS *DispatcherSchedulerSv1) ExecuteActions(args *utils.AttrsExecuteActions, reply *string) error {
+	return dS.dS.SchedulerSv1ExecuteActions(args, reply)
+}
+
+// ExecuteActionPlans execute multiple actionPlans one by one
+func (dS *DispatcherSchedulerSv1) ExecuteActionPlans(args *utils.AttrsExecuteActionPlans, reply *string) (err error) {
+	return dS.dS.SchedulerSv1ExecuteActionPlans(args, reply)
+}
+
 func NewDispatcherSv1(dS *dispatchers.DispatcherService) *DispatcherSv1 {
 	return &DispatcherSv1{dS: dS}
 }

apier/v1/schedulers_it_test.go

@@ -38,6 +38,7 @@ var (
 	schedRpc     *rpc.Client
 	schedConfDIR string //run tests for specific configuration
 )
+
 var sTestsSchedFiltered = []func(t *testing.T){
 	testSchedLoadConfig,
 	testSchedInitDataDb,
@@ -48,6 +49,7 @@ var sTestsSchedFiltered = []func(t *testing.T){
 	testSchedVeifyAllAccounts,
 	testSchedVeifyAccount1001,
 	testSchedVeifyAccount1002and1003,
+	testSchedExecuteAction,
 	testSchedStopEngine,
 }
 
@@ -268,7 +270,59 @@ func testSchedVeifyAccount1002and1003(t *testing.T) {
 		t.Errorf("Expecting: %v, received: %v",
 			10, rply)
 	}
+}
 
+func testSchedExecuteAction(t *testing.T) {
+	if !(schedConfDIR == "tutinternal" || schedConfDIR == "tutmysql" || schedConfDIR == "tutmongo") {
+		t.SkipNow()
+	}
+	// set a new ActionPlan
+	var reply1 string
+	if err := schedRpc.Call(utils.APIerSv1SetActionPlan, &AttrSetActionPlan{
+		Id: "CustomAP",
+		ActionPlan: []*AttrActionPlan{
+			&AttrActionPlan{
+				ActionsId: "ACT_TOPUP_RST_10",
+				Time:      utils.MetaHourly,
+				Weight:    20.0},
+		},
+	}, &reply1); err != nil {
+		t.Error("Got error on APIerSv1.SetActionPlan: ", err.Error())
+	} else if reply1 != utils.OK {
+		t.Errorf("Unexpected reply returned: %s", reply1)
+	}
+	var reply string
+	if err := schedRpc.Call(utils.APIerSv1SetAccount, utils.AttrSetAccount{
+		Tenant:       "cgrates.org",
+		Account:      "CustomAccount",
+		ActionPlanID: "CustomAP",
+	}, &reply); err != nil {
+		t.Fatal(err)
+	}
+
+	var acnt *engine.Account
+	attrs := &utils.AttrGetAccount{
+		Tenant:  "cgrates.org",
+		Account: "CustomAccount",
+	}
+	expected := 0.0
+	if err := schedRpc.Call(utils.APIerSv2GetAccount, attrs, &acnt); err != nil {
+		t.Error(err)
+	} else if rply := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); rply != expected {
+		t.Errorf("Expecting: %v, received: %v",
+			expected, rply)
+	}
+
+	if err := schedRpc.Call(utils.SchedulerSv1ExecuteActions, &utils.AttrsExecuteActions{ActionPlanID: "CustomAP"}, &reply); err != nil {
+		t.Error(err)
+	}
+	expected = 10.0
+	if err := schedRpc.Call(utils.APIerSv2GetAccount, attrs, &acnt); err != nil {
+		t.Error(err)
+	} else if rply := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); rply != expected {
+		t.Errorf("Expecting: %v, received: %v",
+			expected, rply)
+	}
 }
 
 func testSchedStopEngine(t *testing.T) {

dispatchers/scheduler.go

@@ -19,6 +19,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package dispatchers
 
 import (
+	"time"
+
 	"github.com/cgrates/cgrates/utils"
 )
 
@@ -64,3 +66,43 @@ func (dS *DispatcherService) SchedulerSv1Reload(args *utils.CGREventWithArgDispa
 	return dS.Dispatch(args.CGREvent, utils.MetaScheduler, routeID,
 		utils.SchedulerSv1Reload, args, reply)
 }
+
+func (dS *DispatcherService) SchedulerSv1ExecuteActions(args *utils.AttrsExecuteActions, reply *string) (err error) {
+	args.Tenant = utils.FirstNonEmpty(args.Tenant, dS.cfg.GeneralCfg().DefaultTenant)
+	if len(dS.cfg.DispatcherSCfg().AttributeSConns) != 0 {
+		if args.ArgDispatcher == nil {
+			return utils.NewErrMandatoryIeMissing(utils.ArgDispatcherField)
+		}
+		if err = dS.authorize(utils.SchedulerSv1ExecuteActions,
+			args.Tenant,
+			args.APIKey, utils.TimePointer(time.Now())); err != nil {
+			return
+		}
+	}
+	var routeID *string
+	if args.ArgDispatcher != nil {
+		routeID = args.ArgDispatcher.RouteID
+	}
+	return dS.Dispatch(&utils.CGREvent{Tenant: args.Tenant}, utils.MetaScheduler, routeID,
+		utils.SchedulerSv1ExecuteActions, args, reply)
+}
+
+func (dS *DispatcherService) SchedulerSv1ExecuteActionPlans(args *utils.AttrsExecuteActionPlans, reply *string) (err error) {
+	args.Tenant = utils.FirstNonEmpty(args.Tenant, dS.cfg.GeneralCfg().DefaultTenant)
+	if len(dS.cfg.DispatcherSCfg().AttributeSConns) != 0 {
+		if args.ArgDispatcher == nil {
+			return utils.NewErrMandatoryIeMissing(utils.ArgDispatcherField)
+		}
+		if err = dS.authorize(utils.SchedulerSv1ExecuteActionPlans,
+			args.Tenant,
+			args.APIKey, utils.TimePointer(time.Now())); err != nil {
+			return
+		}
+	}
+	var routeID *string
+	if args.ArgDispatcher != nil {
+		routeID = args.ArgDispatcher.RouteID
+	}
+	return dS.Dispatch(&utils.CGREvent{Tenant: args.Tenant}, utils.MetaScheduler, routeID,
+		utils.SchedulerSv1ExecuteActionPlans, args, reply)
+}

utils/apitpdata.go

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