diff --git a/cmd/cgr-engine/cgr-engine.go b/cmd/cgr-engine/cgr-engine.go index 2699b5f57..fc5bada55 100644 --- a/cmd/cgr-engine/cgr-engine.go +++ b/cmd/cgr-engine/cgr-engine.go @@ -140,185 +140,6 @@ func startCdrc(internalCdrSChan, internalRaterChan chan rpcclient.RpcClientConne } } -func startSessionS(internalSMGChan, internalRaterChan, internalResourceSChan, internalThresholdSChan, - internalStatSChan, internalSupplierSChan, internalAttrSChan, internalCDRSChan, internalChargerSChan, - internalDispatcherSChan chan rpcclient.RpcClientConnection, server *utils.Server, - dm *engine.DataManager, exitChan chan bool) { - var err error - var ralsConns, resSConns, threshSConns, statSConns, suplSConns, attrSConns, cdrsConn, chargerSConn rpcclient.RpcClientConnection - - intChargerSChan := internalChargerSChan - intRaterChan := internalRaterChan - intResourceSChan := internalResourceSChan - intThresholdSChan := internalThresholdSChan - intStatSChan := internalStatSChan - intSupplierSChan := internalSupplierSChan - intAttrSChan := internalAttrSChan - intCDRSChan := internalCDRSChan - if cfg.DispatcherSCfg().Enabled { - intChargerSChan = internalDispatcherSChan - intRaterChan = internalDispatcherSChan - intResourceSChan = internalDispatcherSChan - intThresholdSChan = internalDispatcherSChan - intStatSChan = internalDispatcherSChan - intSupplierSChan = internalDispatcherSChan - intAttrSChan = internalDispatcherSChan - intCDRSChan = internalDispatcherSChan - } - - if len(cfg.SessionSCfg().ChargerSConns) != 0 { - chargerSConn, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, cfg.TlsCfg().ClientCerificate, - cfg.TlsCfg().CaCertificate, cfg.GeneralCfg().ConnectAttempts, - cfg.GeneralCfg().Reconnects, cfg.GeneralCfg().ConnectTimeout, - cfg.GeneralCfg().ReplyTimeout, cfg.SessionSCfg().ChargerSConns, - intChargerSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", - utils.SessionS, utils.ChargerS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().RALsConns) != 0 { - ralsConns, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, cfg.TlsCfg().ClientCerificate, - cfg.TlsCfg().CaCertificate, cfg.GeneralCfg().ConnectAttempts, - cfg.GeneralCfg().Reconnects, cfg.GeneralCfg().ConnectTimeout, - cfg.GeneralCfg().ReplyTimeout, cfg.SessionSCfg().RALsConns, - intRaterChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to RALs: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().ResSConns) != 0 { - resSConns, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, - cfg.TlsCfg().ClientCerificate, cfg.TlsCfg().CaCertificate, - cfg.GeneralCfg().ConnectAttempts, cfg.GeneralCfg().Reconnects, - cfg.GeneralCfg().ConnectTimeout, cfg.GeneralCfg().ReplyTimeout, - cfg.SessionSCfg().ResSConns, intResourceSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to ResourceS: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().ThreshSConns) != 0 { - threshSConns, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, - cfg.TlsCfg().ClientCerificate, cfg.TlsCfg().CaCertificate, - cfg.GeneralCfg().ConnectAttempts, cfg.GeneralCfg().Reconnects, - cfg.GeneralCfg().ConnectTimeout, cfg.GeneralCfg().ReplyTimeout, - cfg.SessionSCfg().ThreshSConns, intThresholdSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to ThresholdS: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().StatSConns) != 0 { - statSConns, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, - cfg.TlsCfg().ClientCerificate, cfg.TlsCfg().CaCertificate, - cfg.GeneralCfg().ConnectAttempts, cfg.GeneralCfg().Reconnects, - cfg.GeneralCfg().ConnectTimeout, cfg.GeneralCfg().ReplyTimeout, - cfg.SessionSCfg().StatSConns, intStatSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to StatS: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().SupplSConns) != 0 { - suplSConns, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, - cfg.TlsCfg().ClientCerificate, cfg.TlsCfg().CaCertificate, - cfg.GeneralCfg().ConnectAttempts, cfg.GeneralCfg().Reconnects, - cfg.GeneralCfg().ConnectTimeout, cfg.GeneralCfg().ReplyTimeout, - cfg.SessionSCfg().SupplSConns, intSupplierSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to SupplierS: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().AttrSConns) != 0 { - attrSConns, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, - cfg.TlsCfg().ClientCerificate, cfg.TlsCfg().CaCertificate, - cfg.GeneralCfg().ConnectAttempts, cfg.GeneralCfg().Reconnects, - cfg.GeneralCfg().ConnectTimeout, cfg.GeneralCfg().ReplyTimeout, - cfg.SessionSCfg().AttrSConns, intAttrSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to AttributeS: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - if len(cfg.SessionSCfg().CDRsConns) != 0 { - cdrsConn, err = engine.NewRPCPool(rpcclient.POOL_FIRST, - cfg.TlsCfg().ClientKey, - cfg.TlsCfg().ClientCerificate, cfg.TlsCfg().CaCertificate, - cfg.GeneralCfg().ConnectAttempts, cfg.GeneralCfg().Reconnects, - cfg.GeneralCfg().ConnectTimeout, cfg.GeneralCfg().ReplyTimeout, - cfg.SessionSCfg().CDRsConns, intCDRSChan, false) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to RALs: %s", - utils.SessionS, err.Error())) - exitChan <- true - return - } - } - sReplConns, err := sessions.NewSReplConns(cfg.SessionSCfg().SessionReplicationConns, - cfg.GeneralCfg().Reconnects, cfg.GeneralCfg().ConnectTimeout, - cfg.GeneralCfg().ReplyTimeout) - if err != nil { - utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to SMGReplicationConnection error: <%s>", - utils.SessionS, err.Error())) - exitChan <- true - return - } - sm := sessions.NewSessionS(cfg, ralsConns, resSConns, threshSConns, - statSConns, suplSConns, attrSConns, cdrsConn, chargerSConn, - sReplConns, dm, cfg.GeneralCfg().DefaultTimezone) - //start sync session in a separate gorutine - go func() { - if err = sm.ListenAndServe(exitChan); err != nil { - utils.Logger.Err(fmt.Sprintf("<%s> error: %s!", utils.SessionS, err)) - } - }() - // Pass internal connection via BiRPCClient - internalSMGChan <- sm - // Register RPC handler - smgRpc := v1.NewSMGenericV1(sm) - server.RpcRegister(smgRpc) - - ssv1 := v1.NewSessionSv1(sm) // methods with multiple options - if !cfg.DispatcherSCfg().Enabled { - server.RpcRegister(ssv1) - } - // Register BiRpc handlers - if cfg.SessionSCfg().ListenBijson != "" { - for method, handler := range smgRpc.Handlers() { - server.BiRPCRegisterName(method, handler) - } - for method, handler := range ssv1.Handlers() { - server.BiRPCRegisterName(method, handler) - } - server.ServeBiJSON(cfg.SessionSCfg().ListenBijson, sm.OnBiJSONConnect, sm.OnBiJSONDisconnect) - exitChan <- true - } -} - // startERs handles starting of the EventReader Service func startERs(sSChan, dspSChan chan rpcclient.RpcClientConnection, filterSChan chan *engine.FilterS, @@ -1181,7 +1002,6 @@ func main() { // Define internal connections via channels filterSChan := make(chan *engine.FilterS, 1) internalDispatcherSChan := make(chan rpcclient.RpcClientConnection, 1) - internalSMGChan := make(chan rpcclient.RpcClientConnection, 1) internalAnalyzerSChan := make(chan rpcclient.RpcClientConnection, 1) internalGuardianSChan := make(chan rpcclient.RpcClientConnection, 1) internalLoaderSChan := make(chan rpcclient.RpcClientConnection, 1) @@ -1211,7 +1031,8 @@ func main() { apiv1, _ := srvManager.GetService(utils.ApierV1) apiv2, _ := srvManager.GetService(utils.ApierV2) resp, _ := srvManager.GetService(utils.ResponderS) - srvManager.AddService(chS, attrS, chrS, tS, stS, reS, supS, schS, cdrS, rals) + smg := services.NewSessionService() + srvManager.AddService(chS, attrS, chrS, tS, stS, reS, supS, schS, cdrS, rals, smg) internalAttributeSChan := attrS.GetIntenternalChan() internalChargerSChan := chrS.GetIntenternalChan() internalThresholdSChan := tS.GetIntenternalChan() @@ -1225,6 +1046,7 @@ func main() { internalApierV2Chan := apiv2.GetIntenternalChan() internalRaterChan := resp.GetIntenternalChan() internalRALsv1Chan := rals.GetIntenternalChan() + internalSMGChan := smg.GetIntenternalChan() srvManager.StartServices() cacheS := srvManager.GetCacheS() @@ -1262,14 +1084,6 @@ func main() { // Start CDRC components if necessary go startCdrcs(internalCdrSChan, internalRaterChan, internalDispatcherSChan, filterSChan, exitChan) - // Start SM-Generic - if cfg.SessionSCfg().Enabled { - go startSessionS(internalSMGChan, internalRaterChan, internalRsChan, - internalThresholdSChan, internalStatSChan, internalSupplierSChan, - internalAttributeSChan, internalCdrSChan, internalChargerSChan, - internalDispatcherSChan, server, dm, exitChan) - } - if cfg.ERsCfg().Enabled { go startERs(internalSMGChan, internalDispatcherSChan, filterSChan, cfg.GetReloadChan(config.ERsJson), exitChan) diff --git a/config/config.go b/config/config.go index cca2cce11..4b418cc78 100755 --- a/config/config.go +++ b/config/config.go @@ -1182,7 +1182,10 @@ func (cfg *CGRConfig) SupplierSCfg() *SupplierSCfg { return cfg.supplierSCfg } +// SessionSCfg returns the config for SessionS func (cfg *CGRConfig) SessionSCfg() *SessionSCfg { + cfg.lks[SessionSJson].Lock() + defer cfg.lks[SessionSJson].Unlock() return cfg.sessionSCfg } @@ -1528,6 +1531,7 @@ func (cfg *CGRConfig) reloadSection(section string) (err error) { } fallthrough case SessionSJson: + cfg.rldChans[SessionSJson] <- struct{}{} if !fall { break } diff --git a/services/attributes.go b/services/attributes.go index 5b1ccb3b1..359d26785 100644 --- a/services/attributes.go +++ b/services/attributes.go @@ -65,7 +65,7 @@ func (attrS *AttributeService) Start(sp servmanager.ServiceProvider, waitCache b utils.AttributeS, err.Error())) return } - utils.Logger.Info(fmt.Sprintf("<%s> starting <%s> subsystem", utils.ServiceManager, utils.AttributeS)) + utils.Logger.Info(fmt.Sprintf("<%s> starting <%s> subsystem", utils.CoreS, utils.AttributeS)) attrS.rpc = v1.NewAttributeSv1(attrS.attrS) if !sp.GetConfig().DispatcherSCfg().Enabled { sp.GetServer().RpcRegister(attrS.rpc) diff --git a/services/sessions.go b/services/sessions.go new file mode 100644 index 000000000..69b0c73e8 --- /dev/null +++ b/services/sessions.go @@ -0,0 +1,188 @@ +/* +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 +*/ + +package services + +import ( + "fmt" + "sync" + + v1 "github.com/cgrates/cgrates/apier/v1" + "github.com/cgrates/cgrates/servmanager" + "github.com/cgrates/cgrates/sessions" + "github.com/cgrates/cgrates/utils" + "github.com/cgrates/rpcclient" +) + +// NewSessionService returns the Session Service +func NewSessionService() servmanager.Service { + return &SessionService{ + connChan: make(chan rpcclient.RpcClientConnection, 1), + } +} + +// SessionService implements Service interface +type SessionService struct { + sync.RWMutex + sm *sessions.SessionS + rpc *v1.SMGenericV1 + rpcv1 *v1.SessionSv1 + connChan chan rpcclient.RpcClientConnection +} + +// Start should handle the sercive start +func (smg *SessionService) Start(sp servmanager.ServiceProvider, waitCache bool) (err error) { + if smg.IsRunning() { + return fmt.Errorf("service aleady running") + } + + smg.Lock() + defer smg.Unlock() + var ralsConns, resSConns, threshSConns, statSConns, suplSConns, attrConns, cdrsConn, chargerSConn rpcclient.RpcClientConnection + + if chargerSConn, err = sp.GetConnection(utils.ChargerS, sp.GetConfig().SessionSCfg().ChargerSConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.ChargerS, err.Error())) + return + } + + if ralsConns, err = sp.GetConnection(utils.ResponderS, sp.GetConfig().SessionSCfg().RALsConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.ResponderS, err.Error())) + return + } + + if resSConns, err = sp.GetConnection(utils.ResourceS, sp.GetConfig().SessionSCfg().ResSConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.ResourceS, err.Error())) + return + } + + if threshSConns, err = sp.GetConnection(utils.ThresholdS, sp.GetConfig().SessionSCfg().ThreshSConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.ThresholdS, err.Error())) + return + } + + if statSConns, err = sp.GetConnection(utils.StatS, sp.GetConfig().SessionSCfg().StatSConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.StatS, err.Error())) + return + } + + if suplSConns, err = sp.GetConnection(utils.SupplierS, sp.GetConfig().SessionSCfg().SupplSConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.SupplierS, err.Error())) + return + } + + if attrConns, err = sp.GetConnection(utils.AttributeS, sp.GetConfig().SessionSCfg().AttrSConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.AttributeS, err.Error())) + return + } + + if cdrsConn, err = sp.GetConnection(utils.CDRServer, sp.GetConfig().SessionSCfg().CDRsConns); err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", + utils.SessionS, utils.CDRServer, err.Error())) + return + } + + sReplConns, err := sessions.NewSReplConns(sp.GetConfig().SessionSCfg().SessionReplicationConns, + sp.GetConfig().GeneralCfg().Reconnects, sp.GetConfig().GeneralCfg().ConnectTimeout, + sp.GetConfig().GeneralCfg().ReplyTimeout) + if err != nil { + utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to SMGReplicationConnection error: <%s>", + utils.SessionS, err.Error())) + return + } + + smg.sm = sessions.NewSessionS(sp.GetConfig(), ralsConns, resSConns, threshSConns, + statSConns, suplSConns, attrConns, cdrsConn, chargerSConn, + sReplConns, sp.GetDM(), sp.GetConfig().GeneralCfg().DefaultTimezone) + //start sync session in a separate gorutine + go func() { + if err = smg.sm.ListenAndServe(sp.GetExitChan()); err != nil { + utils.Logger.Err(fmt.Sprintf("<%s> error: %s!", utils.SessionS, err)) + } + }() + // Pass internal connection via BiRPCClient + smg.connChan <- smg.sm + // Register RPC handler + smg.rpc = v1.NewSMGenericV1(smg.sm) + sp.GetServer().RpcRegister(smg.rpc) + + smg.rpcv1 = v1.NewSessionSv1(smg.sm) // methods with multiple options + if !sp.GetConfig().DispatcherSCfg().Enabled { + sp.GetServer().RpcRegister(smg.rpcv1) + } + // Register BiRpc handlers + if sp.GetConfig().SessionSCfg().ListenBijson != "" { + for method, handler := range smg.rpc.Handlers() { + sp.GetServer().BiRPCRegisterName(method, handler) + } + for method, handler := range smg.rpcv1.Handlers() { + sp.GetServer().BiRPCRegisterName(method, handler) + } + sp.GetServer().ServeBiJSON(sp.GetConfig().SessionSCfg().ListenBijson, smg.sm.OnBiJSONConnect, smg.sm.OnBiJSONDisconnect) + return fmt.Errorf("<%s> the session ServeBiJSON should not exit", + utils.SessionS) // imposible this should never reach + } + return +} + +// GetIntenternalChan returns the internal connection chanel +func (smg *SessionService) GetIntenternalChan() (conn chan rpcclient.RpcClientConnection) { + return smg.connChan +} + +// Reload handles the change of config +func (smg *SessionService) Reload(sp servmanager.ServiceProvider) (err error) { + return +} + +// Shutdown stops the service +func (smg *SessionService) Shutdown() (err error) { + smg.Lock() + defer smg.Unlock() + if err = smg.sm.Shutdown(); err != nil { + return + } + smg.sm = nil + smg.rpc = nil + smg.rpcv1 = nil + <-smg.connChan + return +} + +// GetRPCInterface returns the interface to register for server +func (smg *SessionService) GetRPCInterface() interface{} { + return smg.sm +} + +// IsRunning returns if the service is running +func (smg *SessionService) IsRunning() bool { + smg.RLock() + defer smg.RUnlock() + return smg != nil && smg.sm != nil +} + +// ServiceName returns the service name +func (smg *SessionService) ServiceName() string { + return utils.SessionS +} diff --git a/servmanager/servmanager.go b/servmanager/servmanager.go index cf8ef4a24..bdab4206a 100644 --- a/servmanager/servmanager.go +++ b/servmanager/servmanager.go @@ -274,6 +274,9 @@ func (srvMngr *ServiceManager) StartServices() (err error) { if srvMngr.GetConfig().RalsCfg().Enabled { go srvMngr.startService(utils.RALService) } + if srvMngr.GetConfig().SessionSCfg().Enabled { + go srvMngr.startService(utils.SessionS) + } // startServer() return } @@ -333,6 +336,10 @@ func (srvMngr *ServiceManager) handleReload() { if err = srvMngr.reloadService(utils.RALService, srvMngr.GetConfig().RalsCfg().Enabled); err != nil { return } + case <-srvMngr.GetConfig().GetReloadChan(config.SessionSJson): + if err = srvMngr.reloadService(utils.SessionS, srvMngr.GetConfig().SessionSCfg().Enabled); err != nil { + return + } } // handle RPC server }