Consider Filter from Fields in cdrc

cdrc/csv.go

@@ -147,8 +147,10 @@ func (self *CsvRecordsProcessor) recordToStoredCdr(record []string, cdrcCfg *con
 				return nil, err
 			}
 			if pass, err := self.filterS.Pass(tenant,
-				cdrcCfg.Filters, csvProvider); err != nil || !pass {
-				continue // Not passes filters, ignore this CDR
+				cdrFldCfg.Filters, csvProvider); err != nil {
+				return nil, err
+			} else if !pass {
+				continue
 			}
 		}
 		if utils.IsSliceMember([]string{utils.KAM_FLATSTORE, utils.OSIPS_FLATSTORE}, self.dfltCdrcCfg.CdrFormat) { // Hardcode some values in case of flatstore

cdrc/csv_it_test.go

@@ -342,3 +342,89 @@ func TestCsvIT3KillEngine(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+// Begin tests for cdrc csv with new filters
+var fileContent1_4 = `accid21;*prepaid;itsyscom.com;1002;086517174963;2013-02-03 19:54:00;62;val_extra3;"";val_extra1
+accid22;*postpaid;itsyscom.com;1001;+4986517174963;2013-02-03 19:54:00;123;val_extra3;"";val_extra1
+accid23;*postpaid;cgrates.org;1002;086517174963;2013-02-03 19:54:00;76;val_extra3;"";val_extra1`
+
+func TestCsvIT4InitConfig(t *testing.T) {
+	var err error
+	csvCfgPath = path.Join(*dataDir, "conf", "samples", "csvwithfieldfilter")
+	if csvCfg, err = config.NewCGRConfigFromFolder(csvCfgPath); err != nil {
+		t.Fatal("Got config error: ", err.Error())
+	}
+}
+
+// InitDb so we can rely on count
+func TestCsvIT4InitCdrDb(t *testing.T) {
+	if err := engine.InitStorDb(csvCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCsvIT4CreateCdrDirs(t *testing.T) {
+	for _, cdrcProfiles := range csvCfg.CdrcProfiles {
+		for _, cdrcInst := range cdrcProfiles {
+			for _, dir := range []string{cdrcInst.CdrInDir, cdrcInst.CdrOutDir} {
+				if err := os.RemoveAll(dir); err != nil {
+					t.Fatal("Error removing folder: ", dir, err)
+				}
+				if err := os.MkdirAll(dir, 0755); err != nil {
+					t.Fatal("Error creating folder: ", dir, err)
+				}
+			}
+		}
+	}
+}
+
+func TestCsvIT4StartEngine(t *testing.T) {
+	if _, err := engine.StopStartEngine(csvCfgPath, *waitRater); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Connect rpc client to rater
+func TestCsvIT4RpcConn(t *testing.T) {
+	var err error
+	cdrcRpc, err = jsonrpc.Dial("tcp", csvCfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
+	if err != nil {
+		t.Fatal("Could not connect to rater: ", err.Error())
+	}
+}
+
+// Scenario out of first .xml config
+func TestCsvIT4HandleCdr2File(t *testing.T) {
+	fileName := "file1.csv"
+	tmpFilePath := path.Join("/tmp", fileName)
+	if err := ioutil.WriteFile(tmpFilePath, []byte(fileContent1_4), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	if err := os.Rename(tmpFilePath, path.Join("/tmp/csvwithfielfilter/csvit2/in", fileName)); err != nil {
+		t.Fatal("Error moving file to processing directory: ", err)
+	}
+}
+
+func TestCsvIT4ProcessedFiles(t *testing.T) {
+	time.Sleep(time.Duration(2**waitRater) * time.Millisecond)
+	if outContent4, err := ioutil.ReadFile("/tmp/csvwithfielfilter/csvit2/out/file1.csv"); err != nil {
+		t.Error(err)
+	} else if fileContent1_4 != string(outContent4) {
+		t.Errorf("Expecting: %q, received: %q", fileContent1_4, string(outContent4))
+	}
+}
+
+func TestCsvIT4AnalyseCDRs(t *testing.T) {
+	var reply []*engine.ExternalCDR
+	if err := cdrcRpc.Call("ApierV2.GetCdrs", utils.RPCCDRsFilter{}, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if len(reply) != 2 {
+		t.Error("Unexpected number of CDRs returned: ", len(reply))
+	}
+}
+
+func TestCsvIT4KillEngine(t *testing.T) {
+	if err := engine.KillEngine(*waitRater); err != nil {
+		t.Error(err)
+	}
+}

cdrc/fwv.go

@@ -161,14 +161,13 @@ func (self *FwvRecordsProcessor) recordToStoredCdr(record string, cdrcCfg *confi
 		duMultiplyFactor = cdrcCfg.DataUsageMultiplyFactor
 	}
 	for _, cdrFldCfg := range cfgFields {
-		if len(cdrcCfg.Filters) != 0 {
-
+		if len(cdrFldCfg.Filters) != 0 {
 			tenant, err := cdrcCfg.Tenant.ParseValue("")
 			if err != nil {
 				return nil, err
 			}
 			if pass, err := self.filterS.Pass(tenant,
-				cdrcCfg.Filters, fwvProvider); err != nil || !pass {
+				cdrFldCfg.Filters, fwvProvider); err != nil || !pass {
 				continue // Not passes filters, ignore this CDR
 			}
 		}

cdrc/partial_cdr.go

@@ -77,7 +77,8 @@ func (prc *PartialRecordsCache) dumpPartialRecords(originID string) {
 			csvWriter := csv.NewWriter(fileOut)
 			csvWriter.Comma = prc.csvSep
 			for _, cdr := range prc.partialRecords[originID].cdrs {
-				expRec, err := cdr.AsExportRecord(prc.partialRecords[originID].cacheDumpFields, prc.httpSkipTlsCheck, nil, prc.roundDecimals, prc.filterS)
+				expRec, err := cdr.AsExportRecord(prc.partialRecords[originID].cacheDumpFields,
+					prc.httpSkipTlsCheck, nil, prc.roundDecimals, prc.filterS)
 				if err != nil {
 					return nil, err
 				}
@@ -103,7 +104,7 @@ func (prc *PartialRecordsCache) postCDR(originID string) {
 			cdr := prc.partialRecords[originID].MergeCDRs()
 			cdr.Partial = false // force completion
 			var reply string
-			if err := prc.cdrs.Call("CdrsV1.ProcessCDR", cdr, &reply); err != nil {
+			if err := prc.cdrs.Call(utils.CdrsV2ProcessCDR, cdr.AsCGREvent(), &reply); err != nil {
 				utils.Logger.Err(fmt.Sprintf("<Cdrc> Failed sending CDR  %+v from partial cache, error: %s", cdr, err.Error()))
 			} else if reply != utils.OK {
 				utils.Logger.Err(fmt.Sprintf("<Cdrc> Received unexpected reply for CDR, %+v, reply: %s", cdr, reply))

cdrc/xml.go

@@ -168,7 +168,7 @@ func (xmlProc *XMLRecordsProcessor) recordToCDR(xmlEntity tree.Res, cdrcCfg *con
 				return nil, err
 			}
 			if pass, err := xmlProc.filterS.Pass(tenant,
-				cdrcCfg.Filters, xmlProvider); err != nil || !pass {
+				cdrFldCfg.Filters, xmlProvider); err != nil || !pass {
 				continue // Not passes filters, ignore this CDR
 			}
 		}