From fc7a8731b276536c08fe4519aa5fe230f996487d Mon Sep 17 00:00:00 2001 From: Trial97 Date: Fri, 4 Oct 2019 12:34:04 +0300 Subject: [PATCH] Updated Freeswitch Agent --- cmd/cgr-engine/cgr-engine.go | 2 +- services/freeswitchagent.go | 65 ++++++++++++++++++------------------ servmanager/servmanager.go | 14 ++++---- 3 files changed, 41 insertions(+), 40 deletions(-) diff --git a/cmd/cgr-engine/cgr-engine.go b/cmd/cgr-engine/cgr-engine.go index 504205fd8..415776bef 100644 --- a/cmd/cgr-engine/cgr-engine.go +++ b/cmd/cgr-engine/cgr-engine.go @@ -710,9 +710,9 @@ func main() { rals.GetResponder(), rals.GetAPIv1(), rals.GetAPIv2(), cdrS, smg, services.NewEventReaderService(cfg, filterSChan, smg.GetIntenternalChan(), internalDispatcherSChan, exitChan), services.NewDNSAgent(cfg, filterSChan, smg.GetIntenternalChan(), internalDispatcherSChan, exitChan), + services.NewFreeswitchAgent(cfg, smg.GetIntenternalChan(), internalDispatcherSChan, exitChan), ) /* - services.NewFreeswitchAgent(), services.NewKamailioAgent(), services.NewAsteriskAgent(), // partial reload services.NewRadiusAgent(), // partial reload diff --git a/services/freeswitchagent.go b/services/freeswitchagent.go index 86a401e39..868918f22 100644 --- a/services/freeswitchagent.go +++ b/services/freeswitchagent.go @@ -23,6 +23,7 @@ import ( "sync" "github.com/cgrates/cgrates/agents" + "github.com/cgrates/cgrates/config" "github.com/cgrates/cgrates/servmanager" "github.com/cgrates/cgrates/sessions" "github.com/cgrates/cgrates/utils" @@ -30,18 +31,30 @@ import ( ) // NewFreeswitchAgent returns the Freeswitch Agent -func NewFreeswitchAgent() servmanager.Service { - return new(FreeswitchAgent) +func NewFreeswitchAgent(cfg *config.CGRConfig, + sSChan, dispatcherChan chan rpcclient.RpcClientConnection, + exitChan chan bool) servmanager.Service { + return &FreeswitchAgent{ + cfg: cfg, + sSChan: sSChan, + dispatcherChan: dispatcherChan, + exitChan: exitChan, + } } // FreeswitchAgent implements Agent interface type FreeswitchAgent struct { sync.RWMutex + cfg *config.CGRConfig + sSChan chan rpcclient.RpcClientConnection + dispatcherChan chan rpcclient.RpcClientConnection + exitChan chan bool + fS *agents.FSsessions } // Start should handle the sercive start -func (fS *FreeswitchAgent) Start(sp servmanager.ServiceProvider, waitCache bool) (err error) { +func (fS *FreeswitchAgent) Start() (err error) { if fS.IsRunning() { return fmt.Errorf("service aleady running") } @@ -51,25 +64,19 @@ func (fS *FreeswitchAgent) Start(sp servmanager.ServiceProvider, waitCache bool) var sS rpcclient.RpcClientConnection var sSInternal bool utils.Logger.Info("Starting FreeSWITCH agent") - if !sp.GetConfig().DispatcherSCfg().Enabled && sp.GetConfig().FsAgentCfg().SessionSConns[0].Address == utils.MetaInternal { + if !fS.cfg.DispatcherSCfg().Enabled && fS.cfg.FsAgentCfg().SessionSConns[0].Address == utils.MetaInternal { sSInternal = true - srvSessionS, has := sp.GetService(utils.SessionS) - if !has { - utils.Logger.Err(fmt.Sprintf("<%s> Failed to find needed subsystem <%s>", - utils.FreeSWITCHAgent, utils.SessionS)) - return utils.ErrNotFound - } - sSIntConn := <-srvSessionS.GetIntenternalChan() - srvSessionS.GetIntenternalChan() <- sSIntConn + sSIntConn := <-fS.sSChan + fS.sSChan <- sSIntConn sS = utils.NewBiRPCInternalClient(sSIntConn.(*sessions.SessionS)) } else { - if sS, err = sp.NewConnection(utils.SessionS, sp.GetConfig().FsAgentCfg().SessionSConns); err != nil { + if sS, err = NewConnection(fS.cfg, fS.sSChan, fS.dispatcherChan, fS.cfg.FsAgentCfg().SessionSConns); err != nil { utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", utils.FreeSWITCHAgent, utils.SessionS, err.Error())) return } } - fS.fS = agents.NewFSsessions(sp.GetConfig().FsAgentCfg(), sS, sp.GetConfig().GeneralCfg().DefaultTimezone) + fS.fS = agents.NewFSsessions(fS.cfg.FsAgentCfg(), sS, fS.cfg.GeneralCfg().DefaultTimezone) if sSInternal { // bidirectional client backwards connection sS.(*utils.BiRPCInternalClient).SetClientConn(fS.fS) var rply string @@ -84,7 +91,7 @@ func (fS *FreeswitchAgent) Start(sp servmanager.ServiceProvider, waitCache bool) go func() { if err := fS.fS.Connect(); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> error: %s!", utils.FreeSWITCHAgent, err)) - sp.GetExitChan() <- true // stop the engine here + fS.exitChan <- true // stop the engine here } }() return @@ -97,22 +104,16 @@ func (fS *FreeswitchAgent) GetIntenternalChan() (conn chan rpcclient.RpcClientCo } // Reload handles the change of config -func (fS *FreeswitchAgent) Reload(sp servmanager.ServiceProvider) (err error) { +func (fS *FreeswitchAgent) Reload() (err error) { var sS rpcclient.RpcClientConnection var sSInternal bool - if !sp.GetConfig().DispatcherSCfg().Enabled && sp.GetConfig().FsAgentCfg().SessionSConns[0].Address == utils.MetaInternal { + if !fS.cfg.DispatcherSCfg().Enabled && fS.cfg.FsAgentCfg().SessionSConns[0].Address == utils.MetaInternal { sSInternal = true - srvSessionS, has := sp.GetService(utils.SessionS) - if !has { - utils.Logger.Err(fmt.Sprintf("<%s> Failed to find needed subsystem <%s>", - utils.FreeSWITCHAgent, utils.SessionS)) - return utils.ErrNotFound - } - sSIntConn := <-srvSessionS.GetIntenternalChan() - srvSessionS.GetIntenternalChan() <- sSIntConn + sSIntConn := <-fS.sSChan + fS.sSChan <- sSIntConn sS = utils.NewBiRPCInternalClient(sSIntConn.(*sessions.SessionS)) } else { - if sS, err = sp.NewConnection(utils.SessionS, sp.GetConfig().FsAgentCfg().SessionSConns); err != nil { + if sS, err = NewConnection(fS.cfg, fS.sSChan, fS.dispatcherChan, fS.cfg.FsAgentCfg().SessionSConns); err != nil { utils.Logger.Crit(fmt.Sprintf("<%s> Could not connect to %s: %s", utils.FreeSWITCHAgent, utils.SessionS, err.Error())) return @@ -138,7 +139,7 @@ func (fS *FreeswitchAgent) Reload(sp servmanager.ServiceProvider) (err error) { go func() { if err := fS.fS.Connect(); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> error: %s!", utils.FreeSWITCHAgent, err)) - sp.GetExitChan() <- true // stop the engine here + fS.exitChan <- true // stop the engine here } }() return @@ -155,11 +156,6 @@ func (fS *FreeswitchAgent) Shutdown() (err error) { return } -// GetRPCInterface returns the interface to register for server -func (fS *FreeswitchAgent) GetRPCInterface() interface{} { - return fS.fS -} - // IsRunning returns if the service is running func (fS *FreeswitchAgent) IsRunning() bool { fS.RLock() @@ -171,3 +167,8 @@ func (fS *FreeswitchAgent) IsRunning() bool { func (fS *FreeswitchAgent) ServiceName() string { return utils.FreeSWITCHAgent } + +// ShouldRun returns if the service should be running +func (fS *FreeswitchAgent) ShouldRun() bool { + return fS.cfg.FsAgentCfg().Enabled +} diff --git a/servmanager/servmanager.go b/servmanager/servmanager.go index 8eedd8946..31b15aaca 100644 --- a/servmanager/servmanager.go +++ b/servmanager/servmanager.go @@ -276,10 +276,10 @@ func (srvMngr *ServiceManager) StartServices() (err error) { } if srvMngr.GetConfig().DNSAgentCfg().Enabled { go srvMngr.startService(utils.DNSAgent) + } + if srvMngr.GetConfig().FsAgentCfg().Enabled { + go srvMngr.startService(utils.FreeSWITCHAgent) } /* - if srvMngr.GetConfig().FsAgentCfg().Enabled { - go srvMngr.startService(utils.FreeSWITCHAgent) - } if srvMngr.GetConfig().KamAgentCfg().Enabled { go srvMngr.startService(utils.KamailioAgent) } @@ -379,11 +379,11 @@ func (srvMngr *ServiceManager) handleReload() { case <-srvMngr.GetConfig().GetReloadChan(config.DNSAgentJson): if err = srvMngr.reloadService(utils.DNSAgent); err != nil { return + } + case <-srvMngr.GetConfig().GetReloadChan(config.FreeSWITCHAgentJSN): + if err = srvMngr.reloadService(utils.FreeSWITCHAgent); err != nil { + return } /* - case <-srvMngr.GetConfig().GetReloadChan(config.FreeSWITCHAgentJSN): - if err = srvMngr.reloadService(utils.FreeSWITCHAgent); err != nil { - return - } case <-srvMngr.GetConfig().GetReloadChan(config.KamailioAgentJSN): if err = srvMngr.reloadService(utils.KamailioAgent); err != nil { return