Handle *any as subsystem in dispatchers sorting, weight based sorting implementation

dispatchers/dispatchers.go

@@ -75,10 +75,20 @@ func (dS *DispatcherService) dispatcherForEvent(ev *utils.CGREvent,
 	for prflID := range prflIDs {
 		prfl, err := dS.dm.GetDispatcherProfile(ev.Tenant, prflID, true, true, utils.NonTransactional)
 		if err != nil {
-			if err == utils.ErrNotFound {
-				continue
+			if err != utils.ErrNotFound {
+				return nil, err
+			}
+			anyIdxPrfx := utils.ConcatenatedKey(ev.Tenant, utils.META_ANY)
+			if idxKeyPrfx == anyIdxPrfx {
+				continue // already checked *any
+			}
+			// check *any as subsystem
+			if prfl, err = dS.dm.GetDispatcherProfile(ev.Tenant, prflID, true, true, utils.NonTransactional); err != nil {
+				if err == utils.ErrNotFound {
+					continue
+				}
+				return nil, err
 			}
-			return nil, err
 		}
 		if prfl.ActivationInterval != nil && ev.Time != nil &&
 			!prfl.ActivationInterval.IsActiveAtTime(*ev.Time) { // not active

dispatchers/libdispatcher.go

@@ -20,45 +20,11 @@ package dispatchers
 
 import (
 	"fmt"
-	"sort"
 
 	"github.com/cgrates/cgrates/engine"
 	"github.com/cgrates/cgrates/utils"
 )
 
-type DispatcherConn struct {
-	ID        string
-	FilterIDs []string
-	Weight    float64                // applied in case of multiple connections need to be ordered
-	Params    map[string]interface{} // additional parameters stored for a session
-	Blocker   bool                   // no connection after this one
-}
-
-// DispatcherProfile is the config for one Dispatcher
-type DispatcherProfile struct {
-	Tenant             string
-	ID                 string
-	Subsystems         []string
-	FilterIDs          []string
-	ActivationInterval *utils.ActivationInterval // activation interval
-	Strategy           string
-	StrategyParams     map[string]interface{} // ie for distribution, set here the pool weights
-	Weight             float64                // used for profile sorting on match
-	Connections        []*DispatcherConn      // dispatch to these connections
-}
-
-func (dP *DispatcherProfile) TenantID() string {
-	return utils.ConcatenatedKey(dP.Tenant, dP.ID)
-}
-
-// DispatcherProfiles is a sortable list of Dispatcher profiles
-type DispatcherProfiles []*DispatcherProfile
-
-// Sort is part of sort interface, sort based on Weight
-func (dps DispatcherProfiles) Sort() {
-	sort.Slice(dps, func(i, j int) bool { return dps[i].Weight > dps[j].Weight })
-}
-
 // Dispatcher is responsible for routing requests to pool of connections
 // there will be different implementations based on strategy
 type Dispatcher interface {
@@ -71,6 +37,7 @@ type Dispatcher interface {
 
 // newDispatcher constructs instances of Dispatcher
 func newDispatcher(pfl *engine.DispatcherProfile) (d Dispatcher, err error) {
+	pfl.Conns.Sort() // make sure the connections are sorted
 	switch pfl.Strategy {
 	case utils.MetaWeight:
 		d = &WeightDispatcher{pfl: pfl}
@@ -80,15 +47,23 @@ func newDispatcher(pfl *engine.DispatcherProfile) (d Dispatcher, err error) {
 	return
 }
 
+// WeightDispatcher selects the next connection based on weight
 type WeightDispatcher struct {
-	pfl *engine.DispatcherProfile
+	pfl         *engine.DispatcherProfile
+	nextConnIdx int // last returned connection index
 }
 
 func (wd *WeightDispatcher) SetProfile(pfl *engine.DispatcherProfile) {
+	pfl.Conns.Sort()
 	wd.pfl = pfl
 	return
 }
 
 func (wd *WeightDispatcher) NextConnID() (connID string) {
+	connID = wd.pfl.Conns[wd.nextConnIdx].ID
+	wd.nextConnIdx++
+	if wd.nextConnIdx > len(wd.pfl.Conns)-1 {
+		wd.nextConnIdx = 0 // start from beginning
+	}
 	return
 }

engine/dispatcherprfl.go

@@ -24,29 +24,6 @@ import (
 	"github.com/cgrates/cgrates/utils"
 )
 
-// // DispatcherProfile is the config for one Dispatcher
-// type DispatcherProfile struct {
-// 	Tenant             string
-// 	ID                 string
-// 	FilterIDs          []string
-// 	ActivationInterval *utils.ActivationInterval // Activation interval
-// 	Strategy           string
-// 	Hosts              []string // perform data aliasing based on these Attributes
-// 	Weight             float64
-// }
-
-// func (dP *DispatcherProfile) TenantID() string {
-// 	return utils.ConcatenatedKey(dP.Tenant, dP.ID)
-// }
-
-// // ChargerProfiles is a sortable list of Charger profiles
-// type DispatcherProfiles []*DispatcherProfile
-
-// // Sort is part of sort interface, sort based on Weight
-// func (dps DispatcherProfiles) Sort() {
-// 	sort.Slice(dps, func(i, j int) bool { return dps[i].Weight > dps[j].Weight })
-// }
-
 type DispatcherConn struct {
 	ID        string
 	FilterIDs []string
@@ -55,6 +32,13 @@ type DispatcherConn struct {
 	Blocker   bool                   // no connection after this one
 }
 
+type DispatcherConns []*DispatcherConn
+
+// Sort is part of sort interface, sort based on Weight
+func (dConns DispatcherConns) Sort() {
+	sort.Slice(dConns, func(i, j int) bool { return dConns[i].Weight > dConns[j].Weight })
+}
+
 // DispatcherProfile is the config for one Dispatcher
 type DispatcherProfile struct {
 	Tenant             string
@@ -65,7 +49,7 @@ type DispatcherProfile struct {
 	Strategy           string
 	StrategyParams     map[string]interface{} // ie for distribution, set here the pool weights
 	Weight             float64                // used for profile sorting on match
-	Connections        []*DispatcherConn      // dispatch to these connections
+	Conns              DispatcherConns        // dispatch to these connections
 }
 
 func (dP *DispatcherProfile) TenantID() string {