EEs - adding support for AttributeS processing within exporters

config/config.go

@@ -979,8 +979,6 @@ func (cfg *CGRConfig) ERsCfg() *ERsCfg {
 
 // EEsCfg reads the EventExporter configuration
 func (cfg *CGRConfig) EEsCfg() *EEsCfg {
-	cfg.lks[EEsJson].RLock()
-	defer cfg.lks[EEsJson].RUnlock()
 	return cfg.eesCfg
 }
 
@@ -1117,6 +1115,22 @@ func (cfg *CGRConfig) unlockSections() {
 	}
 }
 
+// RLocks will read-lock locks with IDs.
+// User needs to know what he is doing since this can panic
+func (cfg *CGRConfig) RLocks(lkIDs ...string) {
+	for _, lkID := range lkIDs {
+		cfg.lks[lkID].RLock()
+	}
+}
+
+// RLocks will read-lock locks with IDs.
+// User needs to know what he is doing since this can panic
+func (cfg *CGRConfig) RUnlocks(lkIDs ...string) {
+	for _, lkID := range lkIDs {
+		cfg.lks[lkID].RUnlock()
+	}
+}
+
 // V1ReloadConfigFromPath reloads the configuration
 func (cfg *CGRConfig) V1ReloadConfigFromPath(args *ConfigReloadWithArgDispatcher, reply *string) (err error) {
 	if missing := utils.MissingStructFields(args, []string{"Path"}); len(missing) != 0 {

ees/ees.go

@@ -78,26 +78,38 @@ func (eeS *EEService) Shutdown() (err error) {
 }
 
 // ProcessEvent will be called each time a new event is received from readers
-func (eeS *EEService) V1ProcessEvent(cgrEv *utils.CGREvent) (err error) {
-	/*
-		var rplyEv AttrSProcessEventReply
-		attrArgs := &engine.AttrArgsProcessEvent{
-			Context: utils.StringPointer(utils.FirstNonEmpty(
-				utils.IfaceAsString(cgrEv.Opts[utils.Context]),
-				utils.MetaCDRs)),
-			CGREvent:      cgrEv.CGREvent,
-			ArgDispatcher: cgrEv.ArgDispatcher,
-		}
-		if err = cdrS.connMgr.Call(cdrS.cgrCfg.CdrsCfg().AttributeSConns, nil,
-			utils.AttributeSv1ProcessEvent,
-			attrArgs, &rplyEv); err == nil && len(rplyEv.AlteredFields) != 0 {
-			cgrEv.CGREvent = rplyEv.CGREvent
-			cgrEv.Opts = rplyEv.Opts
-		} else if err.Error() == utils.ErrNotFound.Error() {
-			err = nil // cancel ErrNotFound
-		}
-	*/
+func (eeS *EEService) V1ProcessEvent(cgrEv *utils.CGREventWithOpts) (err error) {
+	eeS.cfg.RLocks(config.EEsJson)
+	defer eeS.cfg.RUnlocks(config.EEsJson)
+
 	for _, eeCfg := range eeS.cfg.EEsCfg().Exporters {
+
+		if eeCfg.AttributeSCtx != utils.META_NONE {
+			var rplyEv engine.AttrSProcessEventReply
+			attrArgs := &engine.AttrArgsProcessEvent{
+				AttributeIDs: eeCfg.AttributeSIDs,
+				Context: utils.StringPointer(
+					utils.FirstNonEmpty(
+						eeCfg.AttributeSCtx,
+						utils.IfaceAsString(cgrEv.Opts[utils.Context]),
+						utils.MetaEEs)),
+				CGREvent:      cgrEv.CGREvent,
+				ArgDispatcher: cgrEv.ArgDispatcher,
+			}
+			if err = eeS.connMgr.Call(
+				eeS.cfg.EEsCfg().AttributeSConns, nil,
+				utils.AttributeSv1ProcessEvent,
+				attrArgs, &rplyEv); err == nil && len(rplyEv.AlteredFields) != 0 {
+				cgrEv.CGREvent = rplyEv.CGREvent
+				cgrEv.Opts = rplyEv.Opts
+			} else if err != nil {
+				if err.Error() != utils.ErrNotFound.Error() {
+					return
+				}
+				err = nil // cancel ErrNotFound
+			}
+		}
+
 		eeS.eesMux.RLock()
 		ee, has := eeS.ees[eeCfg.ID]
 		eeS.eesMux.RUnlock()
@@ -107,7 +119,7 @@ func (eeS *EEService) V1ProcessEvent(cgrEv *utils.CGREvent) (err error) {
 			}
 			eeS.ees[eeCfg.ID] = ee
 		}
-		if err = ee.ExportEvent(cgrEv); err != nil {
+		if err = ee.ExportEvent(cgrEv.CGREvent); err != nil {
 			return
 		}
 	}

utils/consts.go

@@ -379,6 +379,7 @@ const (
 	MetaCaches                  = "*caches"
 	MetaCache                   = "*cache"
 	MetaGuardian                = "*guardians"
+	MetaEEs                     = "*ees"
 	MetaContinue                = "*continue"
 	Migrator                    = "migrator"
 	UnsupportedMigrationTask    = "unsupported migration task"