Rename dispatcher to dispatchers in registrarc

This commit is contained in:
andronache
2021-06-10 13:20:22 +03:00
committed by Dan Christian Bogos
parent 213787ee80
commit a27c22ba4e
14 changed files with 113 additions and 142 deletions

View File

@@ -53,7 +53,7 @@ func TestCgrEngineFlags(t *testing.T) {
if err := cgrEngineFlags.Parse([]string{"-httprof_path", "http://example.com/"}); err != nil {
t.Fatal(err)
} else if *httpPprofPath!= "http://example.com/" {
} else if *httpPprofPath != "http://example.com/" {
t.Errorf("Expected http://example.com/, received %+v", *httpPprofPath)
}
@@ -77,7 +77,7 @@ func TestCgrEngineFlags(t *testing.T) {
if err := cgrEngineFlags.Parse([]string{"-memprof_nrfiles", "3"}); err != nil {
t.Fatal(err)
} else if *memProfNrFiles!= 3 {
} else if *memProfNrFiles != 3 {
t.Errorf("Expected 3, received %+v", *memProfNrFiles)
}
@@ -116,4 +116,4 @@ func TestCgrEngineFlags(t *testing.T) {
} else if *preload != "TestPreloadID" {
t.Errorf("Expected 7, received %+v", *preload)
}
}
}

View File

@@ -35,9 +35,9 @@ import (
)
var (
dataDir = flag.String("data_dir", "/usr/share/cgrates", "CGR data dir path here")
dbType = flag.String("dbtype", utils.MetaInternal, "The type of DataBase (Internal/Mongo/mySql)")
encoding = flag.String("rpc", utils.MetaJSON, "what encoding whould be used for rpc comunication")
dataDir = flag.String("data_dir", "/usr/share/cgrates", "CGR data dir path here")
dbType = flag.String("dbtype", utils.MetaInternal, "The type of DataBase (Internal/Mongo/mySql)")
encoding = flag.String("rpc", utils.MetaJSON, "what encoding whould be used for rpc comunication")
waitRater = flag.Int("wait_rater", 100, "Number of miliseconds to wait for rater to start and cache")
)

View File

@@ -1,4 +1,4 @@
// +build integration
// +build integration
/*
Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
@@ -20,28 +20,28 @@ along with this program. If not, see <http://www.gnu.org/licenses/>
package main
import (
"bytes"
"errors"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
"net/rpc"
"net/rpc/jsonrpc"
"os/exec"
"path"
"reflect"
"sort"
"testing"
"time"
)
import (
"bytes"
"errors"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
"net/rpc"
"net/rpc/jsonrpc"
"os/exec"
"path"
"reflect"
"sort"
"testing"
"time"
)
var (
cgrLdrCfgPath string
cgrLdrCfgDir string
cgrLdrCfg *config.CGRConfig
cgrLdrRPC *rpc.Client
cgrLdrTests = []func(t *testing.T) {
cgrLdrCfgDir string
cgrLdrCfg *config.CGRConfig
cgrLdrRPC *rpc.Client
cgrLdrTests = []func(t *testing.T){
testCgrLdrInitCfg,
testCgrLdrInitDataDB,
testCgrLdrInitStorDB,
@@ -68,7 +68,7 @@ var (
)
func TestCGRLoaderRemove(t *testing.T) {
switch *dbType{
switch *dbType {
case utils.MetaInternal:
cgrLdrCfgDir = "tutinternal"
case utils.MetaMongo:
@@ -106,7 +106,7 @@ func testCgrLdrInitStorDB(t *testing.T) {
}
}
func testCgrLdrStartEngine(t *testing.T){
func testCgrLdrStartEngine(t *testing.T) {
if _, err := engine.StartEngine(cgrLdrCfgPath, *waitRater); err != nil {
t.Fatal(err)
}
@@ -198,8 +198,8 @@ func testCgrLdrGetSubsystemsNotLoadedLoad(t *testing.T) {
if err := cgrLdrRPC.Call(utils.APIerSv1GetChargerProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "Raw"},
&replyChrgr); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Errorf("Expected %+v, received %+v", utils.ErrNotFound.Error(), err.Error())
}
t.Errorf("Expected %+v, received %+v", utils.ErrNotFound.Error(), err.Error())
}
}
@@ -220,17 +220,17 @@ func testCgrLdrLoadData(t *testing.T) {
func testCgrLdrGetAttributeProfileAfterLoad(t *testing.T) {
extAttrPrf := &engine.AttributeProfile{
Tenant: "cgrates.org",
ID: "ATTR_ACNT_1001",
Contexts: []string{utils.MetaSessionS},
Tenant: "cgrates.org",
ID: "ATTR_ACNT_1001",
Contexts: []string{utils.MetaSessionS},
FilterIDs: []string{"FLTR_ACCOUNT_1001"},
Weight: 10,
Weight: 10,
Attributes: []*engine.Attribute{
{
FilterIDs: []string{},
Path: "*req.OfficeGroup",
Type: utils.MetaConstant,
Value: config.NewRSRParsersMustCompile("Marketing", utils.InfieldSep),
Path: "*req.OfficeGroup",
Type: utils.MetaConstant,
Value: config.NewRSRParsersMustCompile("Marketing", utils.InfieldSep),
},
},
}
@@ -253,26 +253,26 @@ func testCgrLdrGetAttributeProfileAfterLoad(t *testing.T) {
func testCgrLdrGetFilterAfterLoad(t *testing.T) {
expFilter := &engine.Filter{
Tenant: "cgrates.org",
ID: "FLTR_1",
ID: "FLTR_1",
Rules: []*engine.FilterRule{
{
Type: utils.MetaString,
Type: utils.MetaString,
Element: "~*req.Account",
Values: []string{"1003","1002"},
Values: []string{"1003", "1002"},
},
{
Type: utils.MetaPrefix,
Type: utils.MetaPrefix,
Element: "~*req.Destination",
Values: []string{"10","20"},
Values: []string{"10", "20"},
},
{
Type: utils.MetaRSR,
Type: utils.MetaRSR,
Element: "~*req.Destination",
Values: []string{"1002"},
Values: []string{"1002"},
},
},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, time.July, 29, 15, 0, 0, 0, time.UTC),
ActivationTime: time.Date(2014, time.July, 29, 15, 0, 0, 0, time.UTC),
},
}
var replyFltr *engine.Filter
@@ -287,12 +287,12 @@ func testCgrLdrGetFilterAfterLoad(t *testing.T) {
func testCgrLdrGetResourceProfileAfterLoad(t *testing.T) {
expREsPrf := &engine.ResourceProfile{
Tenant: "cgrates.org",
ID: "RES_ACNT_1001",
FilterIDs: []string{"FLTR_ACCOUNT_1001"},
Weight: 10,
UsageTTL: time.Hour,
Limit: 1,
Tenant: "cgrates.org",
ID: "RES_ACNT_1001",
FilterIDs: []string{"FLTR_ACCOUNT_1001"},
Weight: 10,
UsageTTL: time.Hour,
Limit: 1,
ThresholdIDs: []string{},
}
var replyRes *engine.ResourceProfile
@@ -308,7 +308,7 @@ func testCgrLdrGetResourceProfileAfterLoad(t *testing.T) {
func testCgrLdrGetResourceAfterLoad(t *testing.T) {
expREsPrf := &engine.Resource{
Tenant: "cgrates.org",
ID: "RES_ACNT_1001",
ID: "RES_ACNT_1001",
Usages: map[string]*engine.ResourceUsage{},
}
var replyRes *engine.Resource
@@ -323,23 +323,22 @@ func testCgrLdrGetResourceAfterLoad(t *testing.T) {
func testCgrLdrGetRouteProfileAfterLoad(t *testing.T) {
expRoutePrf := &engine.RouteProfile{
ID: "ROUTE_ACNT_1001",
Tenant: "cgrates.org",
FilterIDs: []string{"FLTR_ACCOUNT_1001"},
Weight: 10,
Sorting: utils.MetaWeight,
ID: "ROUTE_ACNT_1001",
Tenant: "cgrates.org",
FilterIDs: []string{"FLTR_ACCOUNT_1001"},
Weight: 10,
Sorting: utils.MetaWeight,
SortingParameters: []string{},
Routes: []*engine.Route{
{
ID: "route1",
ID: "route1",
Weight: 20,
},
{
ID: "route2",
ID: "route2",
Weight: 10,
},
},
}
var replyRts *engine.RouteProfile
if err := cgrLdrRPC.Call(utils.APIerSv1GetRouteProfile,
@@ -361,13 +360,13 @@ func testCgrLdrGetRouteProfileAfterLoad(t *testing.T) {
func testCgrLdrGetStatsProfileAfterLoad(t *testing.T) {
expStatsprf := &engine.StatQueueProfile{
Tenant: "cgrates.org",
ID: "Stat_1",
FilterIDs: []string{"FLTR_STAT_1"},
Weight: 30,
Tenant: "cgrates.org",
ID: "Stat_1",
FilterIDs: []string{"FLTR_STAT_1"},
Weight: 30,
QueueLength: 100,
TTL: 10 * time.Second,
MinItems: 0,
TTL: 10 * time.Second,
MinItems: 0,
Metrics: []*engine.MetricWithFilters{
{
MetricID: "*tcd",
@@ -379,7 +378,7 @@ func testCgrLdrGetStatsProfileAfterLoad(t *testing.T) {
MetricID: "*acd",
},
},
Blocker: true,
Blocker: true,
ThresholdIDs: []string{utils.MetaNone},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, time.July, 29, 15, 0, 0, 0, time.UTC),
@@ -412,7 +411,7 @@ func testCgrLdrGetStatQueueAfterLoad(t *testing.T) {
replyStQue := make(map[string]string)
if err := cgrLdrRPC.Call(utils.StatSv1GetQueueStringMetrics,
&utils.TenantIDWithAPIOpts{TenantID: &utils.TenantID{Tenant: "cgrates.org", ID: "Stat_1"}},
&replyStQue);err != nil {
&replyStQue); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(expStatQueue, replyStQue) {
t.Errorf("Expected %+v \n, received %+v", utils.ToJSON(expStatQueue), utils.ToJSON(replyStQue))
@@ -421,15 +420,15 @@ func testCgrLdrGetStatQueueAfterLoad(t *testing.T) {
func testCgrLdrGetThresholdProfileAfterLoad(t *testing.T) {
expThPrf := &engine.ThresholdProfile{
Tenant: "cgrates.org",
ID: "THD_ACNT_1001",
Tenant: "cgrates.org",
ID: "THD_ACNT_1001",
FilterIDs: []string{"FLTR_ACCOUNT_1001"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, time.July, 29, 15, 0, 0, 0, time.UTC),
},
Weight: 10,
MaxHits: -1,
MinHits: 0,
Weight: 10,
MaxHits: -1,
MinHits: 0,
ActionIDs: []string{"TOPUP_MONETARY_10"},
}
var replyThdPrf *engine.ThresholdProfile
@@ -445,8 +444,8 @@ func testCgrLdrGetThresholdProfileAfterLoad(t *testing.T) {
func testCgrLdrGetThresholdAfterLoad(t *testing.T) {
expThPrf := &engine.Threshold{
Tenant: "cgrates.org",
ID: "THD_ACNT_1001",
Hits: 0,
ID: "THD_ACNT_1001",
Hits: 0,
}
var replyThdPrf *engine.Threshold
if err := cgrLdrRPC.Call(utils.ThresholdSv1GetThreshold,
@@ -460,12 +459,12 @@ func testCgrLdrGetThresholdAfterLoad(t *testing.T) {
func testCgrLdrGetChargerProfileAfterLoad(t *testing.T) {
expChPrf := &engine.ChargerProfile{
Tenant: "cgrates.org",
ID: "Raw",
FilterIDs: []string{},
RunID: "raw",
Tenant: "cgrates.org",
ID: "Raw",
FilterIDs: []string{},
RunID: "raw",
AttributeIDs: []string{"*constant:*req.RequestType:*none"},
Weight: 20,
Weight: 20,
}
var replyChrgr *engine.ChargerProfile
@@ -500,13 +499,13 @@ func testCgrLdrKillEngine(t *testing.T) {
}
}
func newRPCClient(cfg *config.ListenCfg) (c *rpc.Client, err error) {
switch *encoding {
case utils.MetaJSON:
return jsonrpc.Dial(utils.TCP, cfg.RPCJSONListen)
case utils.MetaGOB:
return rpc.Dial(utils.TCP, cfg.RPCGOBListen)
default:
return nil, errors.New("UNSUPPORTED_RPC")
}
}
func newRPCClient(cfg *config.ListenCfg) (c *rpc.Client, err error) {
switch *encoding {
case utils.MetaJSON:
return jsonrpc.Dial(utils.TCP, cfg.RPCJSONListen)
case utils.MetaGOB:
return rpc.Dial(utils.TCP, cfg.RPCGOBListen)
default:
return nil, errors.New("UNSUPPORTED_RPC")
}
}

View File

@@ -28,7 +28,7 @@ func TestFlags(t *testing.T) {
if err := cgrMigratorFlags.Parse([]string{"-config_path", "true"}); err != nil {
t.Fatal(err)
} else if *cfgPath != "true" {
t.Errorf( "Expected true received:%v ", *cfgPath)
t.Errorf("Expected true received:%v ", *cfgPath)
}
if err := cgrMigratorFlags.Parse([]string{"-exec", "true"}); err != nil {
t.Fatal(err)

File diff suppressed because one or more lines are too long

View File

@@ -1016,7 +1016,7 @@ func (cfg *CGRConfig) checkConfigSanity() error {
}
}
if cfg.registrarCCfg.Dispatchers.Enabled {
if len(cfg.registrarCCfg.Dispatchers.RegistrarSConns) != 0 {
if len(cfg.registrarCCfg.Dispatchers.Hosts) == 0 {
return fmt.Errorf("<%s> missing dispatcher host IDs", utils.RegistrarC)
}
@@ -1030,9 +1030,6 @@ func (cfg *CGRConfig) checkConfigSanity() error {
}
}
}
if len(cfg.registrarCCfg.Dispatchers.RegistrarSConns) == 0 {
return fmt.Errorf("<%s> missing dispatcher connection IDs", utils.RegistrarC)
}
for _, connID := range cfg.registrarCCfg.Dispatchers.RegistrarSConns {
if connID == utils.MetaInternal {
return fmt.Errorf("<%s> internal connection IDs are not supported", utils.RegistrarC)
@@ -1050,7 +1047,7 @@ func (cfg *CGRConfig) checkConfigSanity() error {
}
}
if cfg.registrarCCfg.RPC.Enabled {
if len(cfg.registrarCCfg.RPC.RegistrarSConns) != 0 {
if len(cfg.registrarCCfg.RPC.Hosts) == 0 {
return fmt.Errorf("<%s> missing RPC host IDs", utils.RegistrarC)
}
@@ -1064,9 +1061,6 @@ func (cfg *CGRConfig) checkConfigSanity() error {
}
}
}
if len(cfg.registrarCCfg.RPC.RegistrarSConns) == 0 {
return fmt.Errorf("<%s> missing RPC connection IDs", utils.RegistrarC)
}
for _, connID := range cfg.registrarCCfg.RPC.RegistrarSConns {
if connID == utils.MetaInternal {
return fmt.Errorf("<%s> internal connection IDs are not supported", utils.RegistrarC)

View File

@@ -1514,7 +1514,7 @@ func TestConfigSanityRegistrarCRPC(t *testing.T) {
cfg.registrarCCfg = &RegistrarCCfgs{
RPC: &RegistrarCCfg{
Enabled: true,
RegistrarSConns: []string{utils.MetaLocalHost},
Hosts: map[string][]*RemoteHost{
"hosts": {},
},
@@ -1547,11 +1547,10 @@ func TestConfigSanityRegistrarCRPC(t *testing.T) {
}
cfg.registrarCCfg.RPC.Hosts["hosts"][0].Transport = utils.MetaJSON
expected = "<RegistrarC> missing RPC connection IDs"
expected = "<RegistrarC> connection with id: <*localhost> unsupported transport <*json>"
if err := cfg.CheckConfigSanity(); err == nil || err.Error() != expected {
t.Errorf("Expecting: %+q received: %+q", expected, err)
}
cfg.registrarCCfg.RPC.RegistrarSConns = []string{utils.MetaInternal}
expected = "<RegistrarC> internal connection IDs are not supported"
if err := cfg.CheckConfigSanity(); err == nil || err.Error() != expected {
@@ -1584,7 +1583,7 @@ func TestConfigSanityRegistrarCDispatcher(t *testing.T) {
cfg.registrarCCfg = &RegistrarCCfgs{
Dispatchers: &RegistrarCCfg{
Enabled: true,
RegistrarSConns: []string{utils.MetaLocalHost},
Hosts: map[string][]*RemoteHost{
"hosts": {},
},
@@ -1617,7 +1616,7 @@ func TestConfigSanityRegistrarCDispatcher(t *testing.T) {
}
cfg.registrarCCfg.Dispatchers.Hosts["hosts"][0].Transport = utils.MetaJSON
expected = "<RegistrarC> missing dispatcher connection IDs"
expected = "<RegistrarC> connection with id: <*localhost> unsupported transport <*json>"
if err := cfg.CheckConfigSanity(); err == nil || err.Error() != expected {
t.Errorf("Expecting: %+q received: %+q", expected, err)
}

View File

@@ -537,7 +537,6 @@ type DispatcherSJsonCfg struct {
}
type RegistrarCJsonCfg struct {
Enabled *bool
Registrars_conns *[]string
Hosts map[string][]*RemoteHostJson
Refresh_interval *string

View File

@@ -58,7 +58,6 @@ func (dps RegistrarCCfgs) Clone() (cln *RegistrarCCfgs) {
// RegistrarCCfg is the configuration of registrarc
type RegistrarCCfg struct {
Enabled bool
RegistrarSConns []string
Hosts map[string][]*RemoteHost
RefreshInterval time.Duration
@@ -68,9 +67,6 @@ func (dps *RegistrarCCfg) loadFromJSONCfg(jsnCfg *RegistrarCJsonCfg) (err error)
if jsnCfg == nil {
return nil
}
if jsnCfg.Enabled != nil {
dps.Enabled = *jsnCfg.Enabled
}
if jsnCfg.Registrars_conns != nil {
dps.RegistrarSConns = make([]string, len(*jsnCfg.Registrars_conns))
copy(dps.RegistrarSConns, *jsnCfg.Registrars_conns)
@@ -95,7 +91,6 @@ func (dps *RegistrarCCfg) loadFromJSONCfg(jsnCfg *RegistrarCJsonCfg) (err error)
// AsMapInterface returns the config as a map[string]interface{}
func (dps *RegistrarCCfg) AsMapInterface() (initialMP map[string]interface{}) {
initialMP = map[string]interface{}{
utils.EnabledCfg: dps.Enabled,
utils.RegistrarsConnsCfg: dps.RegistrarSConns,
utils.RefreshIntervalCfg: dps.RefreshInterval.String(),
}
@@ -119,7 +114,6 @@ func (dps *RegistrarCCfg) AsMapInterface() (initialMP map[string]interface{}) {
// Clone returns a deep copy of DispatcherHCfg
func (dps RegistrarCCfg) Clone() (cln *RegistrarCCfg) {
cln = &RegistrarCCfg{
Enabled: dps.Enabled,
RefreshInterval: dps.RefreshInterval,
Hosts: make(map[string][]*RemoteHost),
}

View File

@@ -28,7 +28,6 @@ import (
func TestDispatcherHCfgloadFromJsonCfg(t *testing.T) {
jsonCfg := &RegistrarCJsonCfgs{
RPC: &RegistrarCJsonCfg{
Enabled: utils.BoolPointer(true),
Registrars_conns: &[]string{"*conn1", "*conn2"},
Hosts: map[string][]*RemoteHostJson{
utils.MetaDefault: {
@@ -57,7 +56,6 @@ func TestDispatcherHCfgloadFromJsonCfg(t *testing.T) {
Refresh_interval: utils.StringPointer("5"),
},
Dispatchers: &RegistrarCJsonCfg{
Enabled: utils.BoolPointer(true),
Registrars_conns: &[]string{"*conn1", "*conn2"},
Hosts: map[string][]*RemoteHostJson{
utils.MetaDefault: {
@@ -88,7 +86,6 @@ func TestDispatcherHCfgloadFromJsonCfg(t *testing.T) {
}
expected := &RegistrarCCfgs{
RPC: &RegistrarCCfg{
Enabled: true,
RegistrarSConns: []string{"*conn1", "*conn2"},
Hosts: map[string][]*RemoteHost{
utils.MetaDefault: {
@@ -117,7 +114,6 @@ func TestDispatcherHCfgloadFromJsonCfg(t *testing.T) {
RefreshInterval: 5,
},
Dispatchers: &RegistrarCCfg{
Enabled: true,
RegistrarSConns: []string{"*conn1", "*conn2"},
Hosts: map[string][]*RemoteHost{
utils.MetaDefault: {
@@ -158,7 +154,6 @@ func TestDispatcherHCfgAsMapInterface(t *testing.T) {
cfgJSONStr := `{
"registrarc":{
"rpc":{
"enabled": true,
"registrars_conns": ["*conn1","*conn2"],
"hosts": {
"*default": [
@@ -199,7 +194,6 @@ func TestDispatcherHCfgAsMapInterface(t *testing.T) {
}`
eMap := map[string]interface{}{
utils.RPCCfg: map[string]interface{}{
utils.EnabledCfg: true,
utils.RegistrarsConnsCfg: []string{"*conn1", "*conn2"},
utils.HostsCfg: map[string][]map[string]interface{}{
utils.MetaDefault: {
@@ -216,7 +210,6 @@ func TestDispatcherHCfgAsMapInterface(t *testing.T) {
utils.RefreshIntervalCfg: "0",
},
utils.DispatcherCfg: map[string]interface{}{
utils.EnabledCfg: true,
utils.RegistrarsConnsCfg: []string{"*conn1", "*conn2"},
utils.HostsCfg: map[string][]map[string]interface{}{
utils.MetaDefault: {
@@ -272,13 +265,11 @@ func TestDispatcherHCfgAsMapInterface2(t *testing.T) {
}`
eMap := map[string]interface{}{
utils.DispatcherCfg: map[string]interface{}{
utils.EnabledCfg: false,
utils.RegistrarsConnsCfg: []string{},
utils.HostsCfg: map[string][]map[string]interface{}{},
utils.RefreshIntervalCfg: "5m0s",
},
utils.RPCCfg: map[string]interface{}{
utils.EnabledCfg: false,
utils.RegistrarsConnsCfg: []string{},
utils.HostsCfg: map[string][]map[string]interface{}{},
utils.RefreshIntervalCfg: "5m0s",
@@ -293,7 +284,6 @@ func TestDispatcherHCfgAsMapInterface2(t *testing.T) {
func TestDispatcherHCfgClone(t *testing.T) {
ban := &RegistrarCCfg{
Enabled: true,
RegistrarSConns: []string{"*conn1", "*conn2"},
Hosts: map[string][]*RemoteHost{
utils.MetaDefault: {

View File

@@ -47,11 +47,11 @@ type RegistrarCService struct {
func (dhS *RegistrarCService) ListenAndServe(stopChan, rldChan <-chan struct{}) {
dTm, rTm := &time.Timer{}, &time.Timer{}
var dTmStarted, rTmStarted bool
if dTmStarted = dhS.cfg.RegistrarCCfg().Dispatchers.Enabled; dTmStarted {
if len(dhS.cfg.RegistrarCCfg().Dispatchers.RegistrarSConns) != 0 {
dTm = time.NewTimer(dhS.cfg.RegistrarCCfg().Dispatchers.RefreshInterval)
dhS.registerDispHosts()
}
if rTmStarted = dhS.cfg.RegistrarCCfg().RPC.Enabled; rTmStarted {
if len(dhS.cfg.RegistrarCCfg().RPC.RegistrarSConns) != 0 {
rTm = time.NewTimer(dhS.cfg.RegistrarCCfg().RPC.RefreshInterval)
dhS.registerRPCHosts()
}
@@ -64,19 +64,19 @@ func (dhS *RegistrarCService) ListenAndServe(stopChan, rldChan <-chan struct{})
if dTmStarted {
dTm.Stop()
}
if dTmStarted = dhS.cfg.RegistrarCCfg().Dispatchers.Enabled; dTmStarted {
if len(dhS.cfg.RegistrarCCfg().Dispatchers.RegistrarSConns) != 0 {
dTm = time.NewTimer(dhS.cfg.RegistrarCCfg().Dispatchers.RefreshInterval)
dhS.registerDispHosts()
}
if rTmStarted = dhS.cfg.RegistrarCCfg().RPC.Enabled; rTmStarted {
if len(dhS.cfg.RegistrarCCfg().RPC.RegistrarSConns) != 0 {
rTm = time.NewTimer(dhS.cfg.RegistrarCCfg().RPC.RefreshInterval)
dhS.registerRPCHosts()
}
case <-stopChan:
if dhS.cfg.RegistrarCCfg().Dispatchers.Enabled {
if len(dhS.cfg.RegistrarCCfg().Dispatchers.RegistrarSConns) != 0 {
dTm.Stop()
}
if dhS.cfg.RegistrarCCfg().RPC.Enabled {
if len(dhS.cfg.RegistrarCCfg().RPC.RegistrarSConns) != 0 {
rTm.Stop()
}
return
@@ -93,11 +93,11 @@ func (dhS *RegistrarCService) ListenAndServe(stopChan, rldChan <-chan struct{})
// Shutdown is called to shutdown the service
func (dhS *RegistrarCService) Shutdown() {
utils.Logger.Info(fmt.Sprintf("<%s> service shutdown initialized", utils.RegistrarC))
if dhS.cfg.RegistrarCCfg().Dispatchers.Enabled {
if len(dhS.cfg.RegistrarCCfg().Dispatchers.RegistrarSConns) != 0 {
unregisterHosts(dhS.connMgr, dhS.cfg.RegistrarCCfg().Dispatchers,
dhS.cfg.GeneralCfg().DefaultTenant, utils.RegistrarSv1UnregisterDispatcherHosts)
}
if dhS.cfg.RegistrarCCfg().RPC.Enabled {
if len(dhS.cfg.RegistrarCCfg().RPC.RegistrarSConns) != 0 {
unregisterHosts(dhS.connMgr, dhS.cfg.RegistrarCCfg().RPC,
dhS.cfg.GeneralCfg().DefaultTenant, utils.RegistrarSv1UnregisterRPCHosts)
}

View File

@@ -45,7 +45,6 @@ func TestDispatcherHostsService(t *testing.T) {
Transport: rpcclient.HTTPjson,
}},
}
cfg.RegistrarCCfg().Dispatchers.Enabled = true
cfg.RegistrarCCfg().Dispatchers.Hosts = map[string][]*config.RemoteHost{
utils.MetaDefault: {
{
@@ -127,8 +126,6 @@ func TestDispatcherHostsService(t *testing.T) {
func TestRegistrarcListenAndServe(t *testing.T) {
//cover purposes only
cfg := config.NewDefaultCGRConfig()
cfg.RegistrarCCfg().Dispatchers.Enabled = true
cfg.RegistrarCCfg().RPC.Enabled = true
regStSrv := NewRegistrarCService(cfg, nil)
stopChan := make(chan struct{}, 1)
rldChan := make(chan struct{}, 1)
@@ -194,9 +191,7 @@ func TestRegisterRPCHosts(t *testing.T) {
func TestRegistrarcListenAndServedTmCDispatcher(t *testing.T) {
//cover purposes only
cfg := config.NewDefaultCGRConfig()
cfg.RegistrarCCfg().Dispatchers.Enabled = true
cfg.RegistrarCCfg().Dispatchers.RefreshInterval = 1
cfg.RegistrarCCfg().RPC.Enabled = true
regStSrv := NewRegistrarCService(cfg, nil)
stopChan := make(chan struct{}, 1)
rldChan := make(chan struct{}, 1)
@@ -211,8 +206,6 @@ func TestRegistrarcListenAndServedTmCDispatcher(t *testing.T) {
func TestRegistrarcListenAndServedTmCRPC(t *testing.T) {
//cover purposes only
cfg := config.NewDefaultCGRConfig()
cfg.RegistrarCCfg().Dispatchers.Enabled = true
cfg.RegistrarCCfg().RPC.Enabled = true
cfg.RegistrarCCfg().RPC.RefreshInterval = 1
regStSrv := NewRegistrarCService(cfg, nil)
stopChan := make(chan struct{}, 1)

View File

@@ -103,5 +103,11 @@ func (dspS *RegistrarCService) ServiceName() string {
// ShouldRun returns if the service should be running
func (dspS *RegistrarCService) ShouldRun() bool {
return dspS.cfg.RegistrarCCfg().Dispatchers.Enabled || dspS.cfg.RegistrarCCfg().RPC.Enabled
if len(dspS.cfg.RegistrarCCfg().RPC.RegistrarSConns) != 0 {
return true
}
if len(dspS.cfg.RegistrarCCfg().Dispatchers.RegistrarSConns) != 0 {
return true
}
return false
}

View File

@@ -89,7 +89,6 @@ func TestDispatcherHReload(t *testing.T) {
if err != nil {
t.Errorf("\nExpecting <nil>,\n Received <%+v>", err)
}
cfg.RegistrarCCfg().Dispatchers.Enabled = false
cfg.GetReloadChan(config.RegistrarCJson) <- struct{}{}
time.Sleep(10 * time.Millisecond)
if srv.IsRunning() {