/*
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 Affero 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 Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see
*/
package services
import (
"sync"
"github.com/cgrates/birpc"
v1 "github.com/cgrates/cgrates/apier/v1"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/cores"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/scheduler"
"github.com/cgrates/cgrates/utils"
)
// NewSchedulerService returns the Scheduler Service
func NewSchedulerService(cfg *config.CGRConfig, dm *DataDBService,
cacheS *engine.CacheS, fltrSChan chan *engine.FilterS,
server *cores.Server, internalSchedulerrSChan chan birpc.ClientConnector,
connMgr *engine.ConnManager, anz *AnalyzerService,
srvDep map[string]*sync.WaitGroup) *SchedulerService {
return &SchedulerService{
connChan: internalSchedulerrSChan,
cfg: cfg,
dm: dm,
cacheS: cacheS,
fltrSChan: fltrSChan,
server: server,
connMgr: connMgr,
anz: anz,
srvDep: srvDep,
}
}
// SchedulerService implements Service interface
type SchedulerService struct {
sync.RWMutex
cfg *config.CGRConfig
dm *DataDBService
cacheS *engine.CacheS
fltrSChan chan *engine.FilterS
server *cores.Server
schS *scheduler.Scheduler
connChan chan birpc.ClientConnector
connMgr *engine.ConnManager
anz *AnalyzerService
srvDep map[string]*sync.WaitGroup
}
// Start should handle the sercive start
func (schS *SchedulerService) Start() error {
if schS.IsRunning() {
return utils.ErrServiceAlreadyRunning
}
<-schS.cacheS.GetPrecacheChannel(utils.CacheActionPlans) // wait for ActionPlans to be cached
fltrS := <-schS.fltrSChan
schS.fltrSChan <- fltrS
dbchan := schS.dm.GetDMChan()
datadb := <-dbchan
dbchan <- datadb
schS.Lock()
defer schS.Unlock()
utils.Logger.Info(" Starting CGRateS Scheduler.")
schS.schS = scheduler.NewScheduler(datadb, schS.cfg, fltrS)
go schS.schS.Loop()
srv, err := engine.NewService(v1.NewSchedulerSv1(schS.cfg, datadb, fltrS))
if err != nil {
return err
}
if !schS.cfg.DispatcherSCfg().Enabled {
schS.server.RpcRegister(srv)
}
schS.connChan <- schS.anz.GetInternalCodec(srv, utils.SchedulerS)
return nil
}
// Reload handles the change of config
func (schS *SchedulerService) Reload() (err error) {
schS.Lock()
schS.schS.Reload()
schS.Unlock()
return
}
// Shutdown stops the service
func (schS *SchedulerService) Shutdown() (err error) {
schS.Lock()
schS.schS.Shutdown()
schS.schS = nil
<-schS.connChan
schS.Unlock()
return
}
// IsRunning returns if the service is running
func (schS *SchedulerService) IsRunning() bool {
schS.RLock()
defer schS.RUnlock()
return schS.schS != nil
}
// ServiceName returns the service name
func (schS *SchedulerService) ServiceName() string {
return utils.SchedulerS
}
// GetScheduler returns the Scheduler
func (schS *SchedulerService) GetScheduler() *scheduler.Scheduler {
schS.RLock()
defer schS.RUnlock()
return schS.schS
}
// ShouldRun returns if the service should be running
func (schS *SchedulerService) ShouldRun() bool {
return schS.cfg.SchedulerCfg().Enabled
}