Added Session as service in ServiceManager

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)

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
 		}

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)

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 <http://www.gnu.org/licenses/>
+*/
+
+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
+}

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
 	}