uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-11 10:06:24 +05:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Move CGREvent from utils to engine"

Browse Source 
This reverts commit 02195c47ed.
This commit is contained in:
ionutboangiu
2024-03-01 12:14:40 -05:00
committed by Dan Christian Bogos
parent 0c99cf1df5
commit 0aea8ac641
321 changed files with 3693 additions and 3695 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ers/amqp.go
View File

@@ -137,7 +137,7 @@ func (rdr *AMQPER) processMessage(msg []byte) error {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return err
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.eventChan
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvChan

2
ers/amqp_it_test.go
View File

@@ -115,7 +115,7 @@ func TestAMQPER(t *testing.T) {
if ev.rdrCfg.ID != "amqp" {
t.Errorf("Expected 'amqp' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/amqpv1.go
View File

@@ -173,7 +173,7 @@ func (rdr *AMQPv1ER) processMessage(msg []byte) (err error) {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/amqpv1_it_test.go
View File

@@ -120,7 +120,7 @@ func TestAMQPERv1(t *testing.T) {
if ev.rdrCfg.ID != "amqpv1" {
t.Errorf("Expected 'amqpv1' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/amqpv1_test.go
View File

@@ -40,7 +40,7 @@ func TestAMQPv1ERProcessMessage(t *testing.T) {
cap: nil,
queueID: "cgrates_cdrs",
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.CGRID: "testCgrId",

14
ers/ers.go
View File

@@ -40,7 +40,7 @@ import (
// erEvent is passed from reader to ERs
type erEvent struct {
cgrEvent *engine.CGREvent
cgrEvent *utils.CGREvent
rdrCfg *config.EventReaderCfg
}
@@ -183,7 +183,7 @@ func (erS *ERService) addReader(rdrID string, cfgIdx int) (err error) {
}
// processEvent will be called each time a new event is received from readers
func (erS *ERService) processEvent(cgrEv *engine.CGREvent,
func (erS *ERService) processEvent(cgrEv *utils.CGREvent,
rdrCfg *config.EventReaderCfg) (err error) {
// log the event created if requested by flags
if rdrCfg.Flags.Has(utils.MetaLog) {
@@ -207,7 +207,7 @@ func (erS *ERService) processEvent(cgrEv *engine.CGREvent,
if reqType == utils.MetaAuthorize ||
reqType == utils.MetaMessage ||
reqType == utils.MetaEvent {
if cgrArgs, err = engine.GetRoutePaginatorFromOpts(cgrEv.APIOpts); err != nil {
if cgrArgs, err = utils.GetRoutePaginatorFromOpts(cgrEv.APIOpts); err != nil {
utils.Logger.Warning(fmt.Sprintf("<%s> args extraction for reader <%s> failed because <%s>",
utils.ERs, rdrCfg.ID, err.Error()))
err = nil // reset the error and continue the processing
@@ -334,12 +334,12 @@ func (erS *ERService) closeAllRdrs() {
}
type erEvents struct {
events []*engine.CGREvent
events []*utils.CGREvent
rdrCfg *config.EventReaderCfg
}
// processPartialEvent process the event as a partial event
func (erS *ERService) processPartialEvent(ev *engine.CGREvent, rdrCfg *config.EventReaderCfg) (err error) {
func (erS *ERService) processPartialEvent(ev *utils.CGREvent, rdrCfg *config.EventReaderCfg) (err error) {
// to identify the event the originID and originHost is used to create the CGRID
orgID, err := ev.FieldAsString(utils.OriginID)
if err == utils.ErrNotFound { // the field is missing ignore the event
@@ -355,7 +355,7 @@ func (erS *ERService) processPartialEvent(ev *engine.CGREvent, rdrCfg *config.Ev
var cgrEvs *erEvents
if !has || evs == nil {
cgrEvs = &erEvents{
events: []*engine.CGREvent{ev},
events: []*utils.CGREvent{ev},
rdrCfg: rdrCfg,
}
} else {
@@ -364,7 +364,7 @@ func (erS *ERService) processPartialEvent(ev *engine.CGREvent, rdrCfg *config.Ev
cgrEvs.rdrCfg = rdrCfg
}
var cgrEv *engine.CGREvent
var cgrEv *utils.CGREvent
if cgrEv, err = mergePartialEvents(cgrEvs.events, cgrEvs.rdrCfg, erS.filterS, // merge the events
erS.cfg.GeneralCfg().DefaultTenant,
erS.cfg.GeneralCfg().DefaultTimezone,

46
ers/ers_it_test.go
View File

@@ -114,7 +114,7 @@ func TestERsProcessEventErr(t *testing.T) {
ID: "",
Type: "",
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
}
@@ -193,7 +193,7 @@ func TestERsListenAndServeRdrEvents(t *testing.T) {
srv.rdrErr = make(chan error, 1)
srv.rdrEvents = make(chan *erEvent, 1)
srv.rdrEvents <- &erEvent{
cgrEvent: &engine.CGREvent{
cgrEvent: &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -438,7 +438,7 @@ func TestERsProcessEvent(t *testing.T) {
},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -467,7 +467,7 @@ func TestERsProcessEvent2(t *testing.T) {
},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -495,7 +495,7 @@ func TestERsProcessEvent3(t *testing.T) {
utils.MetaEvent: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
APIOpts: map[string]any{
utils.OptsRoutesLimit: true,
},
@@ -522,7 +522,7 @@ func TestERsProcessEvent4(t *testing.T) {
utils.MetaAuthorize: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
APIOpts: map[string]any{
utils.OptsRoutesLimit: true,
},
@@ -549,7 +549,7 @@ func TestERsProcessEvent5(t *testing.T) {
utils.MetaTerminate: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -580,7 +580,7 @@ func TestERsProcessEvent6(t *testing.T) {
utils.MetaInitiate: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
APIOpts: map[string]any{
utils.OptsRoutesLimit: true,
},
@@ -606,7 +606,7 @@ func TestERsProcessEvent7(t *testing.T) {
utils.MetaUpdate: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
APIOpts: map[string]any{
utils.OptsRoutesLimit: true,
},
@@ -632,7 +632,7 @@ func TestERsProcessEvent8(t *testing.T) {
utils.MetaMessage: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -663,7 +663,7 @@ func TestERsProcessEvent9(t *testing.T) {
utils.MetaCDRs: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -695,7 +695,7 @@ func TestERsProcessEvent10(t *testing.T) {
utils.MetaAccounts: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -752,7 +752,7 @@ func TestERsProcessEvent11(t *testing.T) {
utils.MetaMessage: map[string][]string{},
},
}
cgrEvent := &engine.CGREvent{
cgrEvent := &utils.CGREvent{
Tenant: "",
ID: "",
Time: nil,
@@ -781,7 +781,7 @@ func TestErsOnEvictedMetaDumpToFileOK(t *testing.T) {
val2 := config.NewRSRParsersMustCompile("1001", ",")
val3 := config.NewRSRParsersMustCompile("1002", ",")
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -865,7 +865,7 @@ func TestErsOnEvictedMetaDumpToFileCSVWriteErr(t *testing.T) {
defer os.RemoveAll(dirPath)
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -923,7 +923,7 @@ func TestErsOnEvictedMetaDumpToFileCreateErr(t *testing.T) {
defer os.RemoveAll(dirPath)
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -973,7 +973,7 @@ func TestERsOnEvictedDumpToJSON(t *testing.T) {
}
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -1057,7 +1057,7 @@ func TestERsOnEvictedDumpToJSON(t *testing.T) {
func TestErsOnEvictedDumpToJSONNoPath(t *testing.T) {
dirPath := ""
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -1118,7 +1118,7 @@ func TestErsOnEvictedDumpToJSONMergeError(t *testing.T) {
}
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -1202,7 +1202,7 @@ func TestERsOnEvictedDumpToJSONWithCacheDumpFieldsErrPrefix(t *testing.T) {
}()
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -1274,7 +1274,7 @@ func TestERsOnEvictedDumpToJSONWithCacheDumpFields(t *testing.T) {
}
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -1374,7 +1374,7 @@ func TestErsOnEvictedDumpToJSONInvalidPath(t *testing.T) {
// t.Error(err)
// }
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -1441,7 +1441,7 @@ func TestErsOnEvictedDumpToJSONEncodeErr(t *testing.T) {
t.Error(err)
}
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",

12
ers/ers_test.go
View File

@@ -35,7 +35,7 @@ import (
func TestERsProcessPartialEvent(t *testing.T) {
cfg := config.NewDefaultCGRConfig()
erS := NewERService(cfg, nil, nil)
event := &engine.CGREvent{
event := &utils.CGREvent{
Tenant: "cgrates.org",
ID: "EventERsProcessPartial",
Event: map[string]any{
@@ -80,7 +80,7 @@ func TestErsOnEvictedNilValue(t *testing.T) {
func TestErsOnEvictedMetaPostCDROK(t *testing.T) {
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -134,7 +134,7 @@ func TestErsOnEvictedMetaPostCDRMergeErr(t *testing.T) {
}()
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -194,7 +194,7 @@ func TestErsOnEvictedMetaDumpToFileSetFieldsErr(t *testing.T) {
dirPath := "/tmp/TestErsOnEvictedMetaDumpToFile"
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -250,7 +250,7 @@ func TestErsOnEvictedMetaDumpToFileMergeErr(t *testing.T) {
dirPath := "/tmp/TestErsOnEvictedMetaDumpToFile"
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",
@@ -303,7 +303,7 @@ func TestErsOnEvictedMetaDumpToFileMergeErr(t *testing.T) {
func TestErsOnEvictedMetaDumpToFileEmptyPath(t *testing.T) {
value := &erEvents{
events: []*engine.CGREvent{
events: []*utils.CGREvent{
{
Tenant: "cgrates.org",
ID: "EventErsOnEvicted",

2
ers/filecsv.go
View File

@@ -198,7 +198,7 @@ func (rdr *CSVFileER) processFile(fPath, fName string) (err error) {
utils.ERs, absPath, rowNr, err.Error()))
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/filecsv_it_test.go
View File

@@ -401,7 +401,7 @@ func TestFileCSVProcessEvent(t *testing.T) {
rdrExit: make(chan struct{}),
conReqs: make(chan struct{}, 1),
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.AccountField: "1001",

6
ers/filefwv.go
View File

@@ -220,7 +220,7 @@ func (rdr *FWVFileER) processFile(fPath, fName string) (err error) {
return
}
rdr.offset += rdr.lineLen // increase the offset
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents
@@ -312,7 +312,7 @@ func (rdr *FWVFileER) processTrailer(file *os.File, rowNr, evsPosted int, absPat
utils.ERs, absPath, rowNr, err.Error()))
return err
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents
@@ -357,7 +357,7 @@ func (rdr *FWVFileER) createHeaderMap(record string, rowNr, evsPosted int, absPa
utils.ERs, absPath, rowNr, err.Error()))
return err
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

4
ers/filefwv_it_test.go
View File

@@ -449,7 +449,7 @@ func TestFileFWVProcessTrailer(t *testing.T) {
rdrExit: make(chan struct{}),
conReqs: make(chan struct{}, 1),
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
"OriginID": "testOriginID",
@@ -624,7 +624,7 @@ func TestFileFWVCreateHeaderMap(t *testing.T) {
conReqs: make(chan struct{}, 1),
}
eR.conReqs <- struct{}{}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
"OriginID": "testOriginID",

2
ers/filejson.go
View File

@@ -170,7 +170,7 @@ func (rdr *JSONFileER) processFile(fPath, fName string) (err error) {
utils.ERs, absPath, err.Error()))
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

4
ers/filejson_it_test.go
View File

@@ -346,7 +346,7 @@ func TestFileJSONProcessEvent(t *testing.T) {
rdrExit: make(chan struct{}),
conReqs: make(chan struct{}, 1),
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.AccountField: "account_test",
@@ -363,7 +363,7 @@ func TestFileJSONProcessEvent(t *testing.T) {
},
APIOpts: map[string]any{},
}
// expEvent := &engine.CGREvent{}
// expEvent := &utils.CGREvent{}
eR.conReqs <- struct{}{}
fname := "file1.json"
if err := eR.processFile(filePath, fname); err != nil {

2
ers/filexml.go
View File

@@ -194,7 +194,7 @@ func (rdr *XMLFileER) processFile(fPath, fName string) error {
utils.ERs, absPath, rowNr, err.Error()))
continue
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/filexml_it_test.go
View File

@@ -427,7 +427,7 @@ func TestFileXMLProcessEvent(t *testing.T) {
if err := eR.processFile(filePath, fileName); err != nil {
t.Error(err)
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
"OriginID": "25160047719:0",

2
ers/kafka.go
View File

@@ -193,7 +193,7 @@ func (rdr *KafkaER) processMessage(msg []byte) (err error) {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/kafka_it_test.go
View File

@@ -141,7 +141,7 @@ func TestKafkaER(t *testing.T) {
if ev.rdrCfg.ID != "kafka" {
t.Errorf("expected %s, received %s", "kafka", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/kafka_test.go
View File

@@ -161,7 +161,7 @@ func TestKafkaERProcessMessage(t *testing.T) {
maxWait: time.Duration(1),
cap: make(chan struct{}, 1),
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.ToR: "*voice",

6
ers/libers.go
View File

@@ -30,7 +30,7 @@ import (
)
// mergePartialEvents will unite the events using the reader configuration
func mergePartialEvents(cgrEvs []*engine.CGREvent, cfg *config.EventReaderCfg, fltrS *engine.FilterS, dftTnt, dftTmz, rsrSep string) (cgrEv *engine.CGREvent, err error) {
func mergePartialEvents(cgrEvs []*utils.CGREvent, cfg *config.EventReaderCfg, fltrS *engine.FilterS, dftTnt, dftTmz, rsrSep string) (cgrEv *utils.CGREvent, err error) {
cgrEv = cgrEvs[0] // by default there is at least one event
if len(cgrEvs) != 1 { // need to merge the incoming events
// prepare the field after which the events are ordered
@@ -66,7 +66,7 @@ func mergePartialEvents(cgrEvs []*engine.CGREvent, cfg *config.EventReaderCfg, f
}
// compose the CGREvent from slice
cgrEv = &engine.CGREvent{
cgrEv = &utils.CGREvent{
Tenant: cgrEvs[0].Tenant,
ID: utils.UUIDSha1Prefix(),
Time: utils.TimePointer(time.Now()),
@@ -94,7 +94,7 @@ func mergePartialEvents(cgrEvs []*engine.CGREvent, cfg *config.EventReaderCfg, f
utils.ERs, utils.ToJSON(cgrEv), err.Error()))
return
}
if ev := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant,
if ev := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant,
utils.NestingSep, agReq.Opts); ev != nil { // add the modified fields in the event
for k, v := range ev.Event {
cgrEv.Event[k] = v

2
ers/nats.go
View File

@@ -196,7 +196,7 @@ func (rdr *NatsER) processMessage(msg []byte) (err error) {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/nats_it_test.go
View File

@@ -132,7 +132,7 @@ func testCheckNatsData(t *testing.T, randomCGRID, expData string, ch chan string
if ev.rdrCfg.ID != "nats" {
t.Fatalf("Expected 'nats' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/nats_test.go
View File

@@ -58,7 +58,7 @@ func TestNatsERProcessMessage(t *testing.T) {
rdrErr: make(chan error, 1),
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.ToR: "*voice",

2
ers/s3.go
View File

@@ -134,7 +134,7 @@ func (rdr *S3ER) processMessage(body []byte) (err error) {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/s3_it_test.go
View File

@@ -118,7 +118,7 @@ func TestS3ER(t *testing.T) {
if ev.rdrCfg.ID != "s3" {
t.Errorf("Expected 's3' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/s3_test.go
View File

@@ -83,7 +83,7 @@ func TestS3ERProcessMessage(t *testing.T) {
bucket: "cgrates_cdrs",
session: nil,
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.CGRID: "testCgrId",

2
ers/sql.go
View File

@@ -240,7 +240,7 @@ func (rdr *SQLEventReader) processMessage(msg map[string]any) (err error) {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

8
ers/sql_it_test.go
View File

@@ -247,7 +247,7 @@ func testSQLReader(t *testing.T) {
if ev.rdrCfg.ID != "mysql" {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,
@@ -281,7 +281,7 @@ func testSQLReader2(t *testing.T) {
if ev.rdrCfg.ID != "mysql" {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,
@@ -494,7 +494,7 @@ func testSQLReader3(t *testing.T) {
if ev.rdrCfg.ID != "mysql" {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,
@@ -528,7 +528,7 @@ func testSQLReader4(t *testing.T) {
if ev.rdrCfg.ID != "mysql" {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/sqs.go
View File

@@ -122,7 +122,7 @@ func (rdr *SQSER) processMessage(body []byte) (err error) {
if err = agReq.SetFields(rdr.Config().Fields); err != nil {
return
}
cgrEv := engine.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
cgrEv := utils.NMAsCGREvent(agReq.CGRRequest, agReq.Tenant, utils.NestingSep, agReq.Opts)
rdrEv := rdr.rdrEvents
if _, isPartial := cgrEv.APIOpts[utils.PartialOpt]; isPartial {
rdrEv = rdr.partialEvents

2
ers/sqs_it_test.go
View File

@@ -116,7 +116,7 @@ func TestSQSER(t *testing.T) {
if ev.rdrCfg.ID != "sqs" {
t.Errorf("Expected 'sqs' received `%s`", ev.rdrCfg.ID)
}
expected := &engine.CGREvent{
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Time: ev.cgrEvent.Time,

2
ers/sqs_test.go
View File

@@ -133,7 +133,7 @@ func TestSQSERProcessMessage(t *testing.T) {
queueID: "cgrates_cdrs",
session: nil,
}
expEvent := &engine.CGREvent{
expEvent := &utils.CGREvent{
Tenant: "cgrates.org",
Event: map[string]any{
utils.CGRID: "testCgrId",