Merge branch 'master' of https://github.com/cgrates/cgrates

cmd/cgr-engine/cgr-engine.go

@@ -22,6 +22,13 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"io"
+	"net"
+	"net/rpc"
+	"net/rpc/jsonrpc"
+	"runtime"
+	"time"
+
 	"github.com/cgrates/cgrates/apier/v1"
 	"github.com/cgrates/cgrates/balancer2go"
 	"github.com/cgrates/cgrates/cdrs"
@@ -32,12 +39,6 @@ import (
 	"github.com/cgrates/cgrates/scheduler"
 	"github.com/cgrates/cgrates/sessionmanager"
 	"github.com/cgrates/cgrates/utils"
-	"io"
-	"net"
-	"net/rpc"
-	"net/rpc/jsonrpc"
-	"runtime"
-	"time"
 )
 
 const (
@@ -54,16 +55,16 @@ const (
 )
 
 var (
-	cfgPath  = flag.String("config", "/etc/cgrates/cgrates.cfg", "Configuration file location.")
-	version  = flag.Bool("version", false, "Prints the application version.")
+	cfgPath      = flag.String("config", "/etc/cgrates/cgrates.cfg", "Configuration file location.")
+	version      = flag.Bool("version", false, "Prints the application version.")
 	raterEnabled = flag.Bool("rater", false, "Enforce starting of the rater daemon overwriting config")
 	schedEnabled = flag.Bool("scheduler", false, "Enforce starting of the scheduler daemon overwriting config")
-	bal      = balancer2go.NewBalancer()
-	exitChan = make(chan bool)
-	sm       sessionmanager.SessionManager
-	medi     *mediator.Mediator
-	cfg      *config.CGRConfig
-	err      error
+	bal          = balancer2go.NewBalancer()
+	exitChan     = make(chan bool)
+	sm           sessionmanager.SessionManager
+	medi         *mediator.Mediator
+	cfg          *config.CGRConfig
+	err          error
 )
 
 func listenToRPCRequests(rpcResponder interface{}, apier *apier.ApierV1, rpcAddress string, rpc_encoding string, getter engine.DataStorage, loggerDb engine.LogStorage) {
@@ -207,7 +208,7 @@ func startHistoryScribe() {
 	var scribeServer history.Scribe
 
 	if cfg.HistoryServerEnabled {
-		if scribeServer, err = history.NewFileScribe(cfg.HistoryPath); err != nil {
+		if scribeServer, err = history.NewFileScribe(cfg.HistoryPath, cfg.HistorySavePeriod); err != nil {
 			engine.Logger.Crit(err.Error())
 			exitChan <- true
 			return

config/config.go

@@ -19,9 +19,10 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package config
 
 import (
-	"code.google.com/p/goconf/conf"
 	"errors"
 	"fmt"
+
+	"code.google.com/p/goconf/conf"
 	"github.com/cgrates/cgrates/utils"
 )
 
@@ -102,6 +103,7 @@ type CGRConfig struct {
 	HistoryServer            string   // Address where to reach the master history server: <internal|x.y.z.y:1234>
 	HistoryListen            string   // History server listening interface: <internal|x.y.z.y:1234>
 	HistoryPath              string   // Location on disk where to store history files.
+	HistorySavePeriod        string   // The timout duration between history writes
 }
 
 func (self *CGRConfig) setDefaults() error {
@@ -167,7 +169,7 @@ func (self *CGRConfig) setDefaults() error {
 	self.HistoryServer = "127.0.0.1:2013"
 	self.HistoryListen = "127.0.0.1:2013"
 	self.HistoryPath = "/var/log/cgrates/history"
-
+	self.HistorySavePeriod = "1s"
 	return nil
 }
 
@@ -407,5 +409,8 @@ func loadConfig(c *conf.ConfigFile) (*CGRConfig, error) {
 	if hasOpt = c.HasOption("history_server", "path"); hasOpt {
 		cfg.HistoryPath, _ = c.GetString("history_server", "path")
 	}
+	if hasOpt = c.HasOption("history_server", "save_period"); hasOpt {
+		cfg.HistorySavePeriod, _ = c.GetString("history_server", "save_period")
+	}
 	return cfg, nil
 }

config/config_test.go

@@ -20,9 +20,10 @@ package config
 
 import (
 	"fmt"
-	"github.com/cgrates/cgrates/utils"
 	"reflect"
 	"testing"
+
+	"github.com/cgrates/cgrates/utils"
 )
 
 // Make sure defaults did not change by mistake
@@ -96,6 +97,7 @@ func TestDefaults(t *testing.T) {
 	eCfg.HistoryServerEnabled = false
 	eCfg.HistoryListen = "127.0.0.1:2013"
 	eCfg.HistoryPath = "/var/log/cgrates/history"
+	eCfg.HistorySavePeriod = "1s"
 	if !reflect.DeepEqual(cfg, eCfg) {
 		t.Log(eCfg)
 		t.Log(cfg)
@@ -194,6 +196,7 @@ func TestConfigFromFile(t *testing.T) {
 	eCfg.HistoryServerEnabled = true
 	eCfg.HistoryListen = "test"
 	eCfg.HistoryPath = "test"
+	eCfg.HistorySavePeriod = "test"
 	if !reflect.DeepEqual(cfg, eCfg) {
 		t.Log(eCfg)
 		t.Log(cfg)

config/test_data.txt

@@ -83,3 +83,4 @@ server = test			# Address where to reach the master history server: <internal|x.
 enabled = true			# Starts History service: <true|false>.
 listen = test			# Listening addres for history server: <internal|x.y.z.y:1234>
 path = test				# Location on disk where to store history files.
+save_period = test      # Timeout duration between saves

history/file_scribe.go

@@ -46,9 +46,10 @@ type FileScribe struct {
 	ratingProfiles records
 	loopChecker    chan int
 	waitingFile    string
+	savePeriod     time.Duration
 }
 
-func NewFileScribe(fileRoot string) (*FileScribe, error) {
+func NewFileScribe(fileRoot string, savePeriod string) (*FileScribe, error) {
 	// looking for git
 	gitCommand, err := exec.LookPath("git")
 	if err != nil {
@@ -56,6 +57,9 @@ func NewFileScribe(fileRoot string) (*FileScribe, error) {
 	}
 	s := &FileScribe{fileRoot: fileRoot, gitCommand: gitCommand}
 	s.loopChecker = make(chan int)
+	if s.savePeriod, err = time.ParseDuration(savePeriod); err != nil {
+		return nil, err
+	}
 	s.gitInit()
 	if err := s.load(DESTINATIONS_FILE); err != nil {
 		return nil, err
@@ -88,7 +92,7 @@ func (s *FileScribe) Record(rec *Record, out *int) error {
 	s.waitingFile = fileToSave
 	defer s.Unlock()
 	go func() {
-		t := time.NewTicker(1 * time.Second)
+		t := time.NewTicker(s.savePeriod)
 		select {
 		case <-s.loopChecker:
 			// cancel saving