Add a skelet for fwv ers

cdrc/fwv.go

@@ -23,10 +23,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"net"
 	"os"
-	"strconv"
-	"strings"
 
 	"github.com/cgrates/cgrates/config"
 	"github.com/cgrates/cgrates/engine"
@@ -112,7 +109,7 @@ func (self *FwvRecordsProcessor) ProcessNextRecord() ([]*engine.CDR, error) {
 	}
 	self.processedRecordsNr += 1
 	record := string(buf)
-	fwvProvider := newfwvProvider(record)
+	fwvProvider := config.NewFWVProvider(record, utils.MetaReq)
 	for _, cdrcCfg := range self.cdrcCfgs {
 		tenant, err := cdrcCfg.Tenant.ParseDataProvider(fwvProvider, utils.NestingSep) // each profile of cdrc can have different tenant
 		if err != nil {
@@ -145,8 +142,8 @@ func (self *FwvRecordsProcessor) recordToStoredCdr(record string, cdrcCfg *confi
 	var lazyHttpFields []*config.FCTemplate
 	var cfgFields []*config.FCTemplate
 	var storedCdr *engine.CDR
-	fwvProvider := newfwvProvider(record) // used for filterS and for RSRParsers
-	if self.headerCdr != nil {            // Clone the header CDR so we can use it as base to future processing (inherit fields defined there)
+	fwvProvider := config.NewFWVProvider(record, utils.MetaReq) // used for filterS and for RSRParsers
+	if self.headerCdr != nil {                                  // Clone the header CDR so we can use it as base to future processing (inherit fields defined there)
 		storedCdr = self.headerCdr.Clone()
 	} else {
 		storedCdr = &engine.CDR{OriginHost: "0.0.0.0", ExtraFields: make(map[string]string), Cost: -1}
@@ -248,78 +245,3 @@ func (self *FwvRecordsProcessor) processTrailer() error {
 	}
 	return nil
 }
-
-// newfwvProvider constructs a DataProvider
-func newfwvProvider(record string) (dP config.DataProvider) {
-	dP = &fwvProvider{req: record, cache: config.NewNavigableMap(nil)}
-	return
-}
-
-// fwvProvider implements engine.DataProvider so we can pass it to filters
-type fwvProvider struct {
-	req   string
-	cache *config.NavigableMap
-}
-
-// String is part of engine.DataProvider interface
-// when called, it will display the already parsed values out of cache
-func (fP *fwvProvider) String() string {
-	return utils.ToJSON(fP)
-}
-
-// FieldAsInterface is part of engine.DataProvider interface
-func (fP *fwvProvider) FieldAsInterface(fldPath []string) (data interface{}, err error) {
-	if len(fldPath) == 0 {
-		return
-	}
-	if fldPath[0] != utils.MetaReq || len(fldPath) < 2 {
-		return "", utils.ErrPrefixNotFound(strings.Join(fldPath, utils.NestingSep))
-	}
-	if data, err = fP.cache.FieldAsInterface(fldPath); err == nil ||
-		err != utils.ErrNotFound { // item found in cache
-		return
-	}
-	err = nil // cancel previous err
-	indexes := strings.Split(fldPath[1], "-")
-	if len(indexes) != 2 {
-		return "", fmt.Errorf("Invalid format for index : %+v", fldPath[1])
-	}
-	startIndex, err := strconv.Atoi(indexes[0])
-	if err != nil {
-		return nil, err
-	}
-	if startIndex > len(fP.req) {
-		return "", fmt.Errorf("StartIndex : %+v is greater than : %+v", startIndex, len(fP.req))
-	}
-	finalIndex, err := strconv.Atoi(indexes[1])
-	if err != nil {
-		return nil, err
-	}
-	if finalIndex > len(fP.req) {
-		return "", fmt.Errorf("FinalIndex : %+v is greater than : %+v", finalIndex, len(fP.req))
-	}
-	data = fP.req[startIndex:finalIndex]
-	fP.cache.Set(fldPath, data, false, false)
-	return
-}
-
-// FieldAsString is part of engine.DataProvider interface
-func (fP *fwvProvider) FieldAsString(fldPath []string) (data string, err error) {
-	var valIface interface{}
-	valIface, err = fP.FieldAsInterface(fldPath)
-	if err != nil {
-		return
-	}
-	return utils.IfaceAsString(valIface), nil
-}
-
-// AsNavigableMap is part of engine.DataProvider interface
-func (fP *fwvProvider) AsNavigableMap([]*config.FCTemplate) (
-	nm *config.NavigableMap, err error) {
-	return nil, utils.ErrNotImplemented
-}
-
-// RemoteHost is part of engine.DataProvider interface
-func (fP *fwvProvider) RemoteHost() net.Addr {
-	return utils.LocalAddr()
-}

config/configsanity.go

@@ -429,7 +429,7 @@ func (cfg *CGRConfig) checkConfigSanity() error {
 				if rdr.RunDelay > 0 {
 					return fmt.Errorf("<%s> RunDelay field can not be bigger than zero for reader with ID: %s", utils.ERs, rdr.ID)
 				}
-			case utils.MetaFileXML:
+			case utils.MetaFileXML, utils.MetaFileFWV:
 				for _, dir := range []string{rdr.ProcessedPath, rdr.SourcePath} {
 					if _, err := os.Stat(dir); err != nil && os.IsNotExist(err) {
 						return fmt.Errorf("<%s> Nonexistent folder: %s for reader with ID: %s", utils.ERs, dir, rdr.ID)

config/fwvdp.go

@@ -0,0 +1,109 @@
+/*
+Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
+Copyright (C) ITsysCOM GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>
+*/
+
+package config
+
+import (
+	"fmt"
+	"net"
+	"strconv"
+	"strings"
+
+	"github.com/cgrates/cgrates/utils"
+)
+
+// NewfwvProvider constructs a DataProvider
+func NewFWVProvider(record, pathPrfx string) (dP DataProvider) {
+	dP = &FWVProvider{req: record, cache: NewNavigableMap(nil), pathPrfx: pathPrfx}
+	return
+}
+
+// fwvProvider implements engine.DataProvider so we can pass it to filters
+type FWVProvider struct {
+	req      string
+	cache    *NavigableMap
+	pathPrfx string // if this comes in path it will be ignored
+	// pathPrfx should be reviewed once the cdrc is removed
+}
+
+// String is part of engine.DataProvider interface
+// when called, it will display the already parsed values out of cache
+func (fP *FWVProvider) String() string {
+	return utils.ToJSON(fP)
+}
+
+// FieldAsInterface is part of engine.DataProvider interface
+func (fP *FWVProvider) FieldAsInterface(fldPath []string) (data interface{}, err error) {
+	if len(fldPath) == 0 {
+		return
+	}
+	fwvIdx := fldPath[0]
+	if len(fldPath) == 2 {
+		fwvIdx = fldPath[1]
+	}
+	if fP.pathPrfx != utils.EmptyString && (fldPath[0] != fP.pathPrfx || len(fldPath) < 2) {
+		return "", utils.ErrPrefixNotFound(strings.Join(fldPath, utils.NestingSep))
+	}
+	if data, err = fP.cache.FieldAsInterface(fldPath); err == nil ||
+		err != utils.ErrNotFound { // item found in cache
+		return
+	}
+	err = nil // cancel previous err
+	indexes := strings.Split(fwvIdx, "-")
+	if len(indexes) != 2 {
+		return "", fmt.Errorf("Invalid format for index : %+v", fldPath[1])
+	}
+	startIndex, err := strconv.Atoi(indexes[0])
+	if err != nil {
+		return nil, err
+	}
+	if startIndex > len(fP.req) {
+		return "", fmt.Errorf("StartIndex : %+v is greater than : %+v", startIndex, len(fP.req))
+	}
+	finalIndex, err := strconv.Atoi(indexes[1])
+	if err != nil {
+		return nil, err
+	}
+	if finalIndex > len(fP.req) {
+		return "", fmt.Errorf("FinalIndex : %+v is greater than : %+v", finalIndex, len(fP.req))
+	}
+	data = fP.req[startIndex:finalIndex]
+	fP.cache.Set(fldPath, data, false, false)
+	return
+}
+
+// FieldAsString is part of engine.DataProvider interface
+func (fP *FWVProvider) FieldAsString(fldPath []string) (data string, err error) {
+	var valIface interface{}
+	valIface, err = fP.FieldAsInterface(fldPath)
+	if err != nil {
+		return
+	}
+	return utils.IfaceAsString(valIface), nil
+}
+
+// AsNavigableMap is part of engine.DataProvider interface
+func (fP *FWVProvider) AsNavigableMap([]*FCTemplate) (
+	nm *NavigableMap, err error) {
+	return nil, utils.ErrNotImplemented
+}
+
+// RemoteHost is part of engine.DataProvider interface
+func (fP *FWVProvider) RemoteHost() net.Addr {
+	return utils.LocalAddr()
+}

ers/filefwv.go

@@ -0,0 +1,219 @@
+/*
+Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
+Copyright (C) ITsysCOM GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>
+*/
+
+package ers
+
+import (
+	"bufio"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/cgrates/cgrates/agents"
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
+	"github.com/cgrates/cgrates/utils"
+)
+
+func NewFWVFileERER(cfg *config.CGRConfig, cfgIdx int,
+	rdrEvents chan *erEvent, rdrErr chan error,
+	fltrS *engine.FilterS, rdrExit chan struct{}) (er EventReader, err error) {
+	srcPath := cfg.ERsCfg().Readers[cfgIdx].SourcePath
+	if strings.HasSuffix(srcPath, utils.Slash) {
+		srcPath = srcPath[:len(srcPath)-1]
+	}
+	return &FWVFileER{
+		cgrCfg:    cfg,
+		cfgIdx:    cfgIdx,
+		fltrS:     fltrS,
+		rdrDir:    srcPath,
+		rdrEvents: rdrEvents,
+		rdrError:  rdrErr,
+		rdrExit:   rdrExit}, nil
+}
+
+// XMLFileER implements EventReader interface for .xml files
+type FWVFileER struct {
+	sync.RWMutex
+	cgrCfg        *config.CGRConfig
+	cfgIdx        int // index of config instance within ERsCfg.Readers
+	fltrS         *engine.FilterS
+	rdrDir        string
+	rdrEvents     chan *erEvent // channel to dispatch the events created to
+	rdrError      chan error
+	rdrExit       chan struct{}
+	conReqs       chan struct{} // limit number of opened files
+	lineLen       int64         // Length of the line in the file
+	offset        int64         // Index of the next byte to process
+	trailerOffset int64         // Index where trailer starts, to be used as boundary when reading cdrs
+}
+
+func (rdr *FWVFileER) Config() *config.EventReaderCfg {
+	return rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx]
+}
+
+func (rdr *FWVFileER) Serve() (err error) {
+	switch rdr.Config().RunDelay {
+	case time.Duration(0): // 0 disables the automatic read, maybe done per API
+		return
+	case time.Duration(-1):
+		return watchDir(rdr.rdrDir, rdr.processFile,
+			utils.ERs, rdr.rdrExit)
+	default:
+		go func() {
+			for {
+				// Not automated, process and sleep approach
+				select {
+				case <-rdr.rdrExit:
+					utils.Logger.Info(
+						fmt.Sprintf("<%s> stop monitoring path <%s>",
+							utils.ERs, rdr.rdrDir))
+					return
+				default:
+				}
+				filesInDir, _ := ioutil.ReadDir(rdr.rdrDir)
+				for _, file := range filesInDir {
+					if !strings.HasSuffix(file.Name(), utils.XMLSuffix) { // hardcoded file extension for xml event reader
+						continue // used in order to filter the files from directory
+					}
+					go func(fileName string) {
+						if err := rdr.processFile(rdr.rdrDir, fileName); err != nil {
+							utils.Logger.Warning(
+								fmt.Sprintf("<%s> processing file %s, error: %s",
+									utils.ERs, fileName, err.Error()))
+						}
+					}(file.Name())
+				}
+				time.Sleep(rdr.Config().RunDelay)
+			}
+		}()
+	}
+	return
+}
+
+// processFile is called for each file in a directory and dispatches erEvents from it
+func (rdr *FWVFileER) processFile(fPath, fName string) (err error) {
+	if cap(rdr.conReqs) != 0 { // 0 goes for no limit
+		processFile := <-rdr.conReqs // Queue here for maxOpenFiles
+		defer func() { rdr.conReqs <- processFile }()
+	}
+	absPath := path.Join(fPath, fName)
+	utils.Logger.Info(
+		fmt.Sprintf("<%s> parsing <%s>", utils.ERs, absPath))
+	var file *os.File
+	if file, err = os.Open(absPath); err != nil {
+		return
+	}
+	defer file.Close()
+
+	rowNr := 0 // This counts the rows in the file, not really number of CDRs
+	evsPosted := 0
+	timeStart := time.Now()
+	reqVars := make(map[string]interface{})
+
+	for {
+		cntFld := rdr.Config().ContentFields
+		if rdr.offset == 0 { // First time, set the necessary offsets
+			if err := rdr.setLineLen(file); err != nil {
+				utils.Logger.Err(fmt.Sprintf("<Cdrc> Row 0, error: cannot set lineLen: %s", err.Error()))
+				break
+			}
+			if len(rdr.Config().TrailerFields) != 0 {
+				if fi, err := file.Stat(); err != nil {
+					utils.Logger.Err(fmt.Sprintf("<Cdrc> Row 0, error: cannot get file stats: %s", err.Error()))
+					return err
+				} else {
+					rdr.trailerOffset = fi.Size() - rdr.lineLen
+				}
+			}
+			if len(rdr.Config().HeaderFields) != 0 {
+				buf := make([]byte, rdr.lineLen)
+				if nRead, err := file.Read(buf); err != nil {
+					return err
+				} else if nRead != len(buf) {
+					return fmt.Errorf("In header, line len: %d, have read: %d", rdr.lineLen, nRead)
+				}
+				cntFld = rdr.Config().HeaderFields
+			}
+		}
+		rowNr++ // increment the rowNr after checking if it's not the end of file
+		agReq := agents.NewAgentRequest(
+			config.NewFWVProvider("a", utils.EmptyString), reqVars,
+			nil, nil, rdr.Config().Tenant,
+			rdr.cgrCfg.GeneralCfg().DefaultTenant,
+			utils.FirstNonEmpty(rdr.Config().Timezone,
+				rdr.cgrCfg.GeneralCfg().DefaultTimezone),
+			rdr.fltrS) // create an AgentRequest
+		if pass, err := rdr.fltrS.Pass(agReq.Tenant, rdr.Config().Filters,
+			agReq); err != nil || !pass {
+			continue
+		}
+		navMp, err := agReq.AsNavigableMap(cntFld)
+		if err != nil {
+			utils.Logger.Warning(
+				fmt.Sprintf("<%s> reading file: <%s> row <%d>, ignoring due to error: <%s>",
+					utils.ERs, absPath, rowNr, err.Error()))
+			continue
+		}
+		rdr.rdrEvents <- &erEvent{cgrEvent: navMp.AsCGREvent(
+			agReq.Tenant, utils.NestingSep),
+			rdrCfg: rdr.Config()}
+		evsPosted++
+	}
+
+	if rdr.Config().ProcessedPath != "" {
+		// Finished with file, move it to processed folder
+		outPath := path.Join(rdr.Config().ProcessedPath, fName)
+		if err = os.Rename(absPath, outPath); err != nil {
+			return
+		}
+	}
+
+	utils.Logger.Info(
+		fmt.Sprintf("%s finished processing file <%s>. Total records processed: %d, events posted: %d, run duration: %s",
+			utils.ERs, absPath, rowNr, evsPosted, time.Now().Sub(timeStart)))
+	return
+}
+
+// Sets the line length based on first line, sets offset back to initial after reading
+func (rdr *FWVFileER) setLineLen(file *os.File) error {
+	buff := bufio.NewReader(file)
+	readBytes, err := buff.ReadBytes('\n')
+	if err != nil {
+		return err
+	}
+	rdr.lineLen = int64(len(readBytes))
+	if _, err := file.Seek(0, 0); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (rdr *FWVFileER) processTrailer(file *os.File) error {
+	buf := make([]byte, rdr.lineLen)
+	if nRead, err := file.ReadAt(buf, rdr.trailerOffset); err != nil {
+		return err
+	} else if nRead != len(buf) {
+		return fmt.Errorf("In trailer, line len: %d, have read: %d", rdr.lineLen, nRead)
+	}
+	return nil
+}

ers/reader.go

@@ -42,6 +42,8 @@ func NewEventReader(cfg *config.CGRConfig, cfgIdx int,
 		return NewCSVFileER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
 	case utils.MetaFileXML:
 		return NewXMLFileER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
+	case utils.MetaFileFWV:
+		return NewFWVFileERER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
 	case utils.MetaKafkajsonMap:
 		return NewKafkaER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
 	case utils.MetaSQL: