Added a new configuration option in ers, row_length

config/config_defaults.go

@@ -303,8 +303,9 @@ const CGRATES_CFG_JSON = `
 		{
 			"id": "*default",									// identifier of the EventReader profile
 			"type": "*file_csv",								// reader type <*file_csv>
+			"row_length" : 0, 									// Number of fields from csv file
 			"field_separator": ",",								// separator used in case of csv files
-			"run_delay": "0",										// sleep interval in seconds between consecutive runs, -1 to use automation via inotify or 0 to disable running all together
+			"run_delay": "0",									// sleep interval in seconds between consecutive runs, -1 to use automation via inotify or 0 to disable running all together
 			"concurrent_requests": 1024,						// maximum simultaneous requests/files to process, 0 for unlimited
 			"source_path": "/var/spool/cgrates/cdrc/in",		// read data from this path
 			"processed_path": "/var/spool/cgrates/cdrc/out",	// move processed data here

config/config_it_test.go

@@ -843,6 +843,7 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) {
 				"FailedCallsPrefix":        "",
 				"ID":                       "*default",
 				"ProcessedPath":            "/var/spool/cgrates/cdrc/out",
+				"RowLength":                0,
 				"RunDelay":                 0,
 				"SourcePath":               "/var/spool/cgrates/cdrc/in",
 				"Tenant":                   nil,
@@ -863,6 +864,7 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) {
 				"PartialRecordCache":       0,
 				"ID":                       "file_reader1",
 				"ProcessedPath":            "/tmp/ers/out",
+				"RowLength":                0,
 				"RunDelay":                 -1.,
 				"SourcePath":               "/tmp/ers/in",
 				"Tenant":                   nil,

config/config_json_test.go

@@ -1766,6 +1766,7 @@ func TestDfEventReaderCfg(t *testing.T) {
 			&EventReaderJsonCfg{
 				Id:                  utils.StringPointer(utils.MetaDefault),
 				Type:                utils.StringPointer(utils.MetaFileCSV),
+				Row_length:          utils.IntPointer(0),
 				Field_separator:     utils.StringPointer(","),
 				Run_delay:           utils.StringPointer("0"),
 				Concurrent_requests: utils.IntPointer(1024),

config/erscfg.go

@@ -101,6 +101,7 @@ func (erS *ERsCfg) Clone() (cln *ERsCfg) {
 type EventReaderCfg struct {
 	ID                       string
 	Type                     string
+	RowLength                int
 	FieldSep                 string
 	RunDelay                 time.Duration
 	ConcurrentReqs           int
@@ -128,6 +129,9 @@ func (er *EventReaderCfg) loadFromJsonCfg(jsnCfg *EventReaderJsonCfg, sep string
 	if jsnCfg.Type != nil {
 		er.Type = *jsnCfg.Type
 	}
+	if jsnCfg.Row_length != nil {
+		er.RowLength = *jsnCfg.Row_length
+	}
 	if jsnCfg.Field_separator != nil {
 		er.FieldSep = *jsnCfg.Field_separator
 	}

config/libconfig_json.go

@@ -172,6 +172,7 @@ type ERsJsonCfg struct {
 type EventReaderJsonCfg struct {
 	Id                          *string
 	Type                        *string
+	Row_length                  *int
 	Field_separator             *string
 	Run_delay                   *string
 	Concurrent_requests         *int

docs/ers.rst

@@ -39,6 +39,7 @@ With explanations in the comments:
 			"run_delay":  "-1",			// reading of events it is triggered outside of ERs
 			"field_separator": ";",		// field separator definition
 			"type": "*file_csv",		// type of reader, *file_csv can read .csv files
+			"row_length" : 0,					// Number of fields from csv file
 			"flags": [					// influence processing logic within CGRateS workflow
 				"*cdrs",				//   *cdrs will create CDRs
 				"*log"					//   *log will log the events to syslog

ers/filecsv.go

@@ -130,6 +130,7 @@ func (rdr *CSVFileER) processFile(fPath, fName string) (err error) {
 	}
 	defer file.Close()
 	csvReader := csv.NewReader(bufio.NewReader(file))
+	csvReader.FieldsPerRecord = rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx].RowLength
 	csvReader.Comma = utils.CSV_SEP
 	if len(rdr.Config().FieldSep) > 0 {
 		csvReader.Comma = rune(rdr.Config().FieldSep[0])

ers/flatstore.go

@@ -137,6 +137,7 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) {
 	}
 	defer file.Close()
 	csvReader := csv.NewReader(bufio.NewReader(file))
+	csvReader.FieldsPerRecord = rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx].RowLength
 	csvReader.Comma = ','
 	if len(rdr.Config().FieldSep) > 0 {
 		csvReader.Comma = rune(rdr.Config().FieldSep[0])

ers/partial_csv.go

@@ -144,6 +144,7 @@ func (rdr *PartialCSVFileER) processFile(fPath, fName string) (err error) {
 	}
 	defer file.Close()
 	csvReader := csv.NewReader(bufio.NewReader(file))
+	csvReader.FieldsPerRecord = rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx].RowLength
 	csvReader.Comma = ','
 	if len(rdr.Config().FieldSep) > 0 {
 		csvReader.Comma = rune(rdr.Config().FieldSep[0])