uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-14 20:59:53 +05:00
Code Issues Packages Projects Releases Wiki Activity

Concurent import for CDR files

Browse Source
This commit is contained in:
DanB
2014-07-15 19:46:53 +02:00
parent 430da9b714
commit a722d3242b
1 changed files with 10 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
cdrc/cdrc.go
View File

@@ -150,9 +150,11 @@ func (self *Cdrc) processCdrDir() error {
filesInDir, _ := ioutil.ReadDir(self.cdrInDir)
for _, file := range filesInDir {
if self.cdrType != FS_CSV || path.Ext(file.Name()) != ".csv" {
if err := self.processFile(path.Join(self.cdrInDir, file.Name())); err != nil {
return err
}
go func() { //Enable async processing here
if err := self.processFile(path.Join(self.cdrInDir, file.Name())); err != nil {
engine.Logger.Err(fmt.Sprintf("Processing file %s, error: %s", file, err.Error()))
}
}()
}
}
return nil
@@ -174,9 +176,11 @@ func (self *Cdrc) trackCDRFiles() (err error) {
select {
case ev := <-watcher.Event:
if ev.IsCreate() && (self.cdrType != FS_CSV || path.Ext(ev.Name) != ".csv") {
if err = self.processFile(ev.Name); err != nil {
engine.Logger.Err(fmt.Sprintf("Processing file %s, error: %s", ev.Name, err.Error()))
}
go func() { //Enable async processing here
if err = self.processFile(ev.Name); err != nil {
engine.Logger.Err(fmt.Sprintf("Processing file %s, error: %s", ev.Name, err.Error()))
}
}()
}
case err := <-watcher.Error:
engine.Logger.Err(fmt.Sprintf("Inotify error: %s", err.Error()))