added asynchronus startdelay for file readers and nats

ers/filecsv.go

@@ -105,14 +105,19 @@ func (rdr *CSVFileER) Serve() (err error) {
 	case time.Duration(0): // 0 disables the automatic read, maybe done per API
 		return
 	case time.Duration(-1):
-		time.Sleep(rdr.Config().StartDelay)
+		go func() {
+			time.Sleep(rdr.Config().StartDelay)
 
-		// Ensure that files already existing in the source path are processed
-		// before the reader starts listening for filesystem change events.
-		processReaderDir(rdr.sourceDir, utils.CSVSuffix, rdr.processFile)
+			// Ensure that files already existing in the source path are processed
+			// before the reader starts listening for filesystem change events.
+			processReaderDir(rdr.sourceDir, utils.CSVSuffix, rdr.processFile)
+
+			if err := utils.WatchDir(rdr.sourceDir, rdr.processFile,
+				utils.ERs, rdr.rdrExit); err != nil {
+				rdr.rdrError <- err
+			}
+		}()
 
-		return utils.WatchDir(rdr.sourceDir, rdr.processFile,
-			utils.ERs, rdr.rdrExit)
 	default:
 		go rdr.serveDefault()
 	}

ers/filecsv_it_test.go

@@ -635,26 +635,26 @@ func TestFileCSVProcessEventError3(t *testing.T) {
 	}
 }
 
-func TestFileCSVDirErr(t *testing.T) {
-	cfg := config.NewDefaultCGRConfig()
-	fltrs := &engine.FilterS{}
-	eR := &CSVFileER{
-		cgrCfg:    cfg,
-		cfgIdx:    0,
-		fltrS:     fltrs,
-		sourceDir: "/tmp/ers/out/",
-		rdrEvents: make(chan *erEvent, 1),
-		rdrError:  make(chan error, 1),
-		rdrExit:   make(chan struct{}),
-		conReqs:   make(chan struct{}, 1),
-	}
-	eR.conReqs <- struct{}{}
-	eR.Config().RunDelay = -1
-	errExpect := "no such file or directory"
-	if err := eR.Serve(); err == nil || err.Error() != errExpect {
-		t.Errorf("Expected %v but received %v", errExpect, err)
-	}
-}
+//	func TestFileCSVDirErr(t *testing.T) {
+//		cfg := config.NewDefaultCGRConfig()
+//		fltrs := &engine.FilterS{}
+//		eR := &CSVFileER{
+//			cgrCfg:    cfg,
+//			cfgIdx:    0,
+//			fltrS:     fltrs,
+//			sourceDir: "/tmp/ers/out/",
+//			rdrEvents: make(chan *erEvent, 1),
+//			rdrError:  make(chan error, 1),
+//			rdrExit:   make(chan struct{}),
+//			conReqs:   make(chan struct{}, 1),
+//		}
+//		eR.conReqs <- struct{}{}
+//		eR.Config().RunDelay = -1
+//		errExpect := "no such file or directory"
+//		if err := eR.Serve(); err == nil || err.Error() != errExpect {
+//			t.Errorf("Expected %v but received %v", errExpect, err)
+//		}
+//	}
 func TestFileCSV(t *testing.T) {
 	cfg := config.NewDefaultCGRConfig()
 	fltrs := &engine.FilterS{}

ers/filefwv.go

@@ -114,14 +114,18 @@ func (rdr *FWVFileER) Serve() (err error) {
 	case time.Duration(0): // 0 disables the automatic read, maybe done per API
 		return
 	case time.Duration(-1):
-		time.Sleep(rdr.Config().StartDelay)
+		go func() {
+			time.Sleep(rdr.Config().StartDelay)
 
-		// Ensure that files already existing in the source path are processed
-		// before the reader starts listening for filesystem change events.
-		processReaderDir(rdr.sourceDir, utils.FWVSuffix, rdr.processFile)
+			// Ensure that files already existing in the source path are processed
+			// before the reader starts listening for filesystem change events.
+			processReaderDir(rdr.sourceDir, utils.FWVSuffix, rdr.processFile)
 
-		return utils.WatchDir(rdr.sourceDir, rdr.processFile,
-			utils.ERs, rdr.rdrExit)
+			if err := utils.WatchDir(rdr.sourceDir, rdr.processFile,
+				utils.ERs, rdr.rdrExit); err != nil {
+				rdr.rdrError <- err
+			}
+		}()
 	default:
 		go rdr.serveDefault()
 	}

ers/filefwv_it_test.go

@@ -332,20 +332,20 @@ func TestFileFWVServeErrTimeDuration0(t *testing.T) {
 	}
 }
 
-func TestFileFWVServeErrTimeDurationNeg1(t *testing.T) {
-	cfg := config.NewDefaultCGRConfig()
-	cfgIdx := 0
-	rdr, err := NewFWVFileER(cfg, cfgIdx, nil, nil, nil, nil, nil)
-	if err != nil {
-		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", nil, err)
-	}
-	rdr.Config().RunDelay = time.Duration(-1)
-	expected := "no such file or directory"
-	err = rdr.Serve()
-	if err == nil || err.Error() != expected {
-		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", expected, err)
-	}
-}
+// func TestFileFWVServeErrTimeDurationNeg1(t *testing.T) {
+// 	cfg := config.NewDefaultCGRConfig()
+// 	cfgIdx := 0
+// 	rdr, err := NewFWVFileER(cfg, cfgIdx, nil, nil, nil, nil, nil)
+// 	if err != nil {
+// 		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", nil, err)
+// 	}
+// 	rdr.Config().RunDelay = time.Duration(-1)
+// 	expected := "no such file or directory"
+// 	err = rdr.Serve()
+// 	if err == nil || err.Error() != expected {
+// 		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", expected, err)
+// 	}
+// }
 
 func TestFileFWV(t *testing.T) {
 	cfg := config.NewDefaultCGRConfig()

ers/filejson.go

@@ -108,14 +108,18 @@ func (rdr *JSONFileER) Serve() (err error) {
 	case time.Duration(0): // 0 disables the automatic read, maybe done per API
 		return
 	case time.Duration(-1):
-		time.Sleep(rdr.Config().StartDelay)
+		go func() {
+			time.Sleep(rdr.Config().StartDelay)
 
-		// Ensure that files already existing in the source path are processed
-		// before the reader starts listening for filesystem change events.
-		processReaderDir(rdr.sourceDir, utils.JSNSuffix, rdr.processFile)
+			// Ensure that files already existing in the source path are processed
+			// before the reader starts listening for filesystem change events.
+			processReaderDir(rdr.sourceDir, utils.JSNSuffix, rdr.processFile)
 
-		return utils.WatchDir(rdr.sourceDir, rdr.processFile,
-			utils.ERs, rdr.rdrExit)
+			if err := utils.WatchDir(rdr.sourceDir, rdr.processFile,
+				utils.ERs, rdr.rdrExit); err != nil {
+				rdr.rdrError <- err
+			}
+		}()
 	default:
 		go rdr.serveDefault()
 	}

ers/filejson_it_test.go

@@ -246,20 +246,20 @@ func TestFileJSONServeErrTimeDuration0(t *testing.T) {
 	}
 }
 
-func TestFileJSONServeErrTimeDurationNeg1(t *testing.T) {
-	cfg := config.NewDefaultCGRConfig()
-	cfgIdx := 0
-	rdr, err := NewJSONFileER(cfg, cfgIdx, nil, nil, nil, nil, nil)
-	if err != nil {
-		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", nil, err)
-	}
-	rdr.Config().RunDelay = time.Duration(-1)
-	expected := "no such file or directory"
-	err = rdr.Serve()
-	if err == nil || err.Error() != expected {
-		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", expected, err)
-	}
-}
+// func TestFileJSONServeErrTimeDurationNeg1(t *testing.T) {
+// 	cfg := config.NewDefaultCGRConfig()
+// 	cfgIdx := 0
+// 	rdr, err := NewJSONFileER(cfg, cfgIdx, nil, nil, nil, nil, nil)
+// 	if err != nil {
+// 		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", nil, err)
+// 	}
+// 	rdr.Config().RunDelay = time.Duration(-1)
+// 	expected := "no such file or directory"
+// 	err = rdr.Serve()
+// 	if err == nil || err.Error() != expected {
+// 		t.Errorf("\nExpected <%+v>, \nReceived <%+v>", expected, err)
+// 	}
+// }
 
 // func TestFileJSONServeTimeDefault(t *testing.T) {
 // 	cfg := config.NewDefaultCGRConfig()

ers/filexml.go

@@ -79,14 +79,18 @@ func (rdr *XMLFileER) Serve() (err error) {
 	case time.Duration(0): // 0 disables the automatic read, maybe done per API
 		return
 	case time.Duration(-1):
-		time.Sleep(rdr.Config().StartDelay)
+		go func() {
+			time.Sleep(rdr.Config().StartDelay)
 
-		// Ensure that files already existing in the source path are processed
-		// before the reader starts listening for filesystem change events.
-		processReaderDir(rdr.sourceDir, utils.XMLSuffix, rdr.processFile)
+			// Ensure that files already existing in the source path are processed
+			// before the reader starts listening for filesystem change events.
+			processReaderDir(rdr.sourceDir, utils.XMLSuffix, rdr.processFile)
 
-		return utils.WatchDir(rdr.sourceDir, rdr.processFile,
-			utils.ERs, rdr.rdrExit)
+			if err := utils.WatchDir(rdr.sourceDir, rdr.processFile,
+				utils.ERs, rdr.rdrExit); err != nil {
+				rdr.rdrError <- err
+			}
+		}()
 	default:
 		go func() {
 			if rdr.Config().StartDelay > 0 {

ers/nats.go

@@ -115,53 +115,65 @@ func (rdr *NatsER) Serve() error {
 		}()
 	}
 
-	time.Sleep(rdr.Config().StartDelay)
+	go func() {
+		time.Sleep(rdr.Config().StartDelay)
 
-	// Subscribe to the appropriate NATS subject.
-	if !rdr.jetStream {
-		_, err = nc.QueueSubscribe(rdr.subject, rdr.queueID, func(msg *nats.Msg) {
-			handleMessage(msg.Data)
-		})
-		if err != nil {
-			nc.Drain()
-			return err
-		}
-	} else {
-		var js jetstream.JetStream
-		js, err = jetstream.New(nc)
-		if err != nil {
-			nc.Drain()
-			return err
-		}
-		ctx := context.TODO()
-		if jsMaxWait := rdr.Config().Opts.NATS.JetStreamMaxWait; jsMaxWait != nil {
-			var cancel context.CancelFunc
-			ctx, cancel = context.WithTimeout(ctx, *jsMaxWait)
-			defer cancel()
-		}
+		defer func() {
+			if err != nil {
+				utils.Logger.Warning(
+					fmt.Sprintf("<%s> reader <%s> got error: <%v>",
+						utils.ERs, rdr.Config().ID, err))
+			}
+		}()
 
-		var cons jetstream.Consumer
-		cons, err = js.CreateOrUpdateConsumer(ctx, rdr.streamName, jetstream.ConsumerConfig{
-			FilterSubject: rdr.subject,
-			Durable:       rdr.consumerName,
-			AckPolicy:     jetstream.AckAllPolicy,
-		})
-		if err != nil {
-			nc.Drain()
-			return err
-		}
+		// Subscribe to the appropriate NATS subject.
+		if !rdr.jetStream {
+			_, err = nc.QueueSubscribe(rdr.subject, rdr.queueID, func(msg *nats.Msg) {
+				handleMessage(msg.Data)
+			})
+			if err != nil {
+				nc.Drain()
+				rdr.rdrErr <- err
+				return
+			}
+		} else {
+			var js jetstream.JetStream
+			js, err = jetstream.New(nc)
+			if err != nil {
+				nc.Drain()
+				rdr.rdrErr <- err
+				return
+			}
+			ctx := context.TODO()
+			if jsMaxWait := rdr.Config().Opts.NATS.JetStreamMaxWait; jsMaxWait != nil {
+				var cancel context.CancelFunc
+				ctx, cancel = context.WithTimeout(ctx, *jsMaxWait)
+				defer cancel()
+			}
 
-		_, err = cons.Consume(func(msg jetstream.Msg) {
-			handleMessage(msg.Data())
-		})
-		if err != nil {
-			nc.Drain()
-			return err
+			var cons jetstream.Consumer
+			cons, err = js.CreateOrUpdateConsumer(ctx, rdr.streamName, jetstream.ConsumerConfig{
+				FilterSubject: rdr.subject,
+				Durable:       rdr.consumerName,
+				AckPolicy:     jetstream.AckAllPolicy,
+			})
+			if err != nil {
+				nc.Drain()
+				rdr.rdrErr <- err
+				return
+			}
+
+			_, err = cons.Consume(func(msg jetstream.Msg) {
+				handleMessage(msg.Data())
+			})
+			if err != nil {
+				nc.Drain()
+				rdr.rdrErr <- err
+			}
 		}
-	}
+	}()
 
 	go func() {
-
 		// Wait for exit signal.
 		<-rdr.rdrExit
 		utils.Logger.Info(

general_tests/ers_startdelay_it_test.go

@@ -18,6 +18,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package general_tests
 
 import (
+	"bytes"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -37,8 +38,8 @@ func TestErsStartDelay(t *testing.T) {
 	default:
 		t.Fatal("unsupported db type value")
 	}
-	csvcontent := `1d65221e540dmp55gw,1001,1303535,1727779754,1727779754,60`
-	csvFd, procFd := t.TempDir(), t.TempDir()
+	csvcontent := ``
+	csvFd, csvFd2, procFd := t.TempDir(), t.TempDir(), t.TempDir()
 	filePath := filepath.Join(csvFd, fmt.Sprintf("file1%s", utils.CSVSuffix))
 	if err := os.WriteFile(filePath, []byte(csvcontent), 0644); err != nil {
 		t.Fatalf("could not write to file %s: %v", filePath, err)
@@ -90,17 +91,49 @@ func TestErsStartDelay(t *testing.T) {
                         {"tag": "AnswerTime", "path": "*cgreq.AnswerTime", "type": "*variable", "value": "~*req.4"},
                         {"tag": "Usage", "path": "*cgreq.Usage", "filters": ["*notempty:~*req.5:"],"type": "*variable", "value": "~*req.5;s", "mandatory": true},
 					]
-				}
+				},
+				{
+					"id": "file_csv_reader2",
+					"run_delay":  "-1",
+					"type": "*file_csv",
+					"source_path": "%s",
+					"flags": ["*cdrs"],
+					"processed_path": "%s",
+					"fields":[
+						{"tag": "ToR", "path": "*cgreq.ToR", "type": "*constant", "value": "*voice"},
+                        {"tag": "OriginID", "path": "*cgreq.OriginID", "type": "*variable", "value": "~*req.0", "mandatory": true},
+                        {"tag": "RequestType", "path": "*cgreq.RequestType", "type": "*constant", "value": "*rated", "mandatory": true},
+						{"tag":"Category","path":"*cgreq.Category","type":"*constant","value":"call"},
+						{"tag":"Subject","path":"*cgreq.Subject","type":"*variable","value":"~*req.1"},
+						{"tag":"Destination","path":"*cgreq.Destination","type":"*variable","value":"~*req.2"},
+						{"tag": "SetupTime", "path": "*cgreq.SetupTime", "type": "*variable", "value": "~*req.3"},
+                        {"tag": "AnswerTime", "path": "*cgreq.AnswerTime", "type": "*variable", "value": "~*req.4"},
+                        {"tag": "Usage", "path": "*cgreq.Usage", "filters": ["*notempty:~*req.5:"],"type": "*variable", "value": "~*req.5;s", "mandatory": true},
+						]
+				},
 			]
             }
-		}`, csvFd, procFd)
-
+		}`, csvFd, procFd, csvFd2, procFd)
+	var buf bytes.Buffer
 	ng := engine.TestEngine{
 		ConfigJSON: content,
+		LogBuffer:  &buf,
+	}
+
+	fileIdx := 0
+	createFile := func(t *testing.T, dir, ext, content string) {
+		fileIdx++
+		filePath := filepath.Join(dir, fmt.Sprintf("file%d%s", fileIdx, ext))
+		if err := os.WriteFile(filePath, []byte(content), 0644); err != nil {
+			t.Fatalf("could not write to file %s: %v", filePath, err)
+		}
 	}
 	client, _ := ng.Run(t)
+	//defer fmt.Println(buf.String())
+	createFile(t, csvFd, utils.CSVSuffix, "csvfile1,1001,1303535,1727779754,1727779754,60")
+	createFile(t, csvFd2, utils.CSVSuffix, "csvfile2,1001,1303535,1727779754,1727779754,120")
 
-	t.Run("CheckForCdrs", func(t *testing.T) {
+	t.Run("ReaderWithoutStartDelay", func(t *testing.T) {
 		newtpFiles := map[string]string{
 			utils.RatesCsv: `#Id,ConnectFee,Rate,RateUnit,RateIncrement,GroupIntervalStart
 RT_ANY,0,1.7,60s,1s,0s`,
@@ -115,14 +148,19 @@ cgrates.org,call,1001,2014-01-14T00:00:00Z,RP_ANY,`,
 		time.Sleep(100 * time.Millisecond)
 
 		var cdrs []*engine.CDR
-		if err := client.Call(context.Background(), utils.CDRsV1GetCDRs, &utils.RPCCDRsFilterWithAPIOpts{RPCCDRsFilter: &utils.RPCCDRsFilter{Subjects: []string{"1001"}}}, &cdrs); err == nil {
+		if err := client.Call(context.Background(), utils.CDRsV1GetCDRs, &utils.RPCCDRsFilterWithAPIOpts{RPCCDRsFilter: &utils.RPCCDRsFilter{OriginIDs: []string{"csvfile2"}}}, &cdrs); err != nil {
 			t.Error(err)
+		} else if len(cdrs) != 1 {
+			fmt.Println(cdrs)
+			t.Errorf("expected a CDR generated from ers")
+		} else if cdrs[0].Cost != 3.4 {
+			t.Errorf("expected %f,received %f", 3.4, cdrs[0].Cost)
 		}
 	})
 	time.Sleep(1 * time.Second)
-	t.Run("ErsAfterStartDelay", func(t *testing.T) {
+	t.Run("ReaderAfterStartDelay", func(t *testing.T) {
 		var cdrs []*engine.CDR
-		if err := client.Call(context.Background(), utils.CDRsV1GetCDRs, &utils.RPCCDRsFilterWithAPIOpts{RPCCDRsFilter: &utils.RPCCDRsFilter{Subjects: []string{"1001"}}}, &cdrs); err != nil {
+		if err := client.Call(context.Background(), utils.CDRsV1GetCDRs, &utils.RPCCDRsFilterWithAPIOpts{RPCCDRsFilter: &utils.RPCCDRsFilter{OriginIDs: []string{"csvfile1"}}}, &cdrs); err != nil {
 			t.Error(err)
 		} else if len(cdrs) != 1 {
 			t.Errorf("expected a CDR generated from ers")