Prepared flatStore

2026-02-11 18:16:24 +05:00 · 2021-05-04 18:02:42 +03:00
parent ef60112e6b
commit 3070139cb2
7 changed files with 52 additions and 43 deletions
--- a/ees/sql.go
+++ b/ees/sql.go
@@ -71,7 +71,7 @@ func (sqlEe *SQLEe) NewSQLEeURL(cgrCfg *config.CGRConfig) (dialect gorm.Dialecto
 	password, _ := u.User.Password()

 	dbname := utils.SQLDefaultDBName
-	if vals, has := cgrCfg.EEsCfg().Exporters[sqlEe.cfgIdx].Opts[utils.SQLDBName]; has {
+	if vals, has := cgrCfg.EEsCfg().Exporters[sqlEe.cfgIdx].Opts[utils.SQLDBNameOpt]; has {
 		dbname = utils.IfaceAsString(vals)
 	}
 	ssl := utils.SQLDefaultSSLMode
@@ -79,8 +79,8 @@ func (sqlEe *SQLEe) NewSQLEeURL(cgrCfg *config.CGRConfig) (dialect gorm.Dialecto
 		ssl = utils.IfaceAsString(vals)
 	}
 	// tableName is mandatory in opts
-	if iface, has := cgrCfg.EEsCfg().Exporters[sqlEe.cfgIdx].Opts[utils.SQLTableName]; !has {
-		return nil, utils.NewErrMandatoryIeMissing(utils.SQLTableName)
+	if iface, has := cgrCfg.EEsCfg().Exporters[sqlEe.cfgIdx].Opts[utils.SQLTableNameOpt]; !has {
+		return nil, utils.NewErrMandatoryIeMissing(utils.SQLTableNameOpt)
 	} else {
 		sqlEe.tableName = utils.IfaceAsString(iface)
 	}
--- a/ees/sql_it_test.go
+++ b/ees/sql_it_test.go
@@ -300,8 +300,8 @@ func TestOpenDB3Err(t *testing.T) {

 func TestSQLExportEvent1(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "cgrates"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "cgrates"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = `mysql://cgrates:CGRateS.org@127.0.0.1:3306`
 	cgrEv := new(utils.CGREvent)
 	newIDb := engine.NewInternalDB(nil, nil, true)
@@ -343,8 +343,8 @@ func TestSQLExportEvent1(t *testing.T) {

 func TestSQLExportEvent2(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "cgrates"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "cgrates"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = `mysql://cgrates:CGRateS.org@127.0.0.1:3306`
 	cgrEv := new(utils.CGREvent)
 	newIDb := engine.NewInternalDB(nil, nil, true)
@@ -386,8 +386,8 @@ func TestSQLExportEvent2(t *testing.T) {

 func TestSQLExportEvent3(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "cgrates"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "cgrates"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = `mysql://cgrates:CGRateS.org@127.0.0.1:3306`
 	cgrEv := new(utils.CGREvent)
 	newIDb := engine.NewInternalDB(nil, nil, true)
@@ -429,8 +429,8 @@ func TestSQLExportEvent3(t *testing.T) {

 func TestSQLExportEvent4(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "cgrates"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "cgrates"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = `mysql://cgrates:CGRateS.org@127.0.0.1:3306`
 	cgrEv := new(utils.CGREvent)
 	newIDb := engine.NewInternalDB(nil, nil, true)
--- a/ees/sql_test.go
+++ b/ees/sql_test.go
@@ -60,8 +60,8 @@ func TestSqlGetMetrics(t *testing.T) {

 func TestNewSQLeUrl(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "postgres"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "postgres"
 	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SSLModeCfg] = "test"
 	newIDb := engine.NewInternalDB(nil, nil, true)
 	newDM := engine.NewDataManager(newIDb, cgrCfg.CacheCfg(), nil)
@@ -84,8 +84,8 @@ func TestNewSQLeUrl(t *testing.T) {

 func TestNewSQLeUrlSQL(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "mysql"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "mysql"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = `mysql://cgrates:CGRateS.org@127.0.0.1:3306`
 	newIDb := engine.NewInternalDB(nil, nil, true)
 	newDM := engine.NewDataManager(newIDb, cgrCfg.CacheCfg(), nil)
@@ -110,8 +110,8 @@ func TestNewSQLeUrlSQL(t *testing.T) {

 func TestNewSQLeUrlPostgres(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "postgres"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "postgres"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = `postgres://cgrates:CGRateS.org@127.0.0.1:3306`
 	newIDb := engine.NewInternalDB(nil, nil, true)
 	newDM := engine.NewDataManager(newIDb, cgrCfg.CacheCfg(), nil)
@@ -136,8 +136,8 @@ func TestNewSQLeUrlPostgres(t *testing.T) {

 func TestNewSQLeExportPathError(t *testing.T) {
 	cgrCfg := config.NewDefaultCGRConfig()
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableName] = "expTable"
-	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBName] = "postgres"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLTableNameOpt] = "expTable"
+	cgrCfg.EEsCfg().Exporters[0].Opts[utils.SQLDBNameOpt] = "postgres"
 	cgrCfg.EEsCfg().Exporters[0].ExportPath = ":foo"
 	newIDb := engine.NewInternalDB(nil, nil, true)
 	newDM := engine.NewDataManager(newIDb, cgrCfg.CacheCfg(), nil)
--- a/ers/flatstore.go
+++ b/ers/flatstore.go
@@ -39,6 +39,13 @@ import (
 	"github.com/cgrates/cgrates/utils"
 )

+type fstRecord struct {
+	inv      []string
+	bye      []string
+	ack      []string
+	fileName string
+}
+
 func NewFlatstoreER(cfg *config.CGRConfig, cfgIdx int,
 	rdrEvents chan *erEvent, rdrErr chan error,
 	fltrS *engine.FilterS, rdrExit chan struct{}) (er EventReader, err error) {
--- a/ers/sql.go
+++ b/ers/sql.go
@@ -268,7 +268,7 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string, opts map[string]interfac
 	rdr.connType = u.Scheme

 	dbname := utils.SQLDefaultDBName
-	if vals, has := opts[utils.SQLDBName]; has {
+	if vals, has := opts[utils.SQLDBNameOpt]; has {
 		dbname = utils.IfaceAsString(vals)
 	}
 	ssl := utils.SQLDefaultSSLMode
@@ -277,7 +277,7 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string, opts map[string]interfac
 	}

 	rdr.tableName = utils.CDRsTBL
-	if vals, has := opts[utils.SQLTableName]; has {
+	if vals, has := opts[utils.SQLTableNameOpt]; has {
 		rdr.tableName = utils.IfaceAsString(vals)
 	}
 	switch rdr.connType {
@@ -317,7 +317,7 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string, opts map[string]interfac
 	}

 	outDBname = utils.SQLDefaultDBName
-	if vals, has := processedOpt[utils.SQLDBName]; has {
+	if vals, has := processedOpt[utils.SQLDBNameOpt]; has {
 		outDBname = utils.IfaceAsString(vals)
 	}
 	outSSL = utils.SQLDefaultSSLMode
@@ -325,7 +325,7 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string, opts map[string]interfac
 		outSSL = utils.IfaceAsString(vals)
 	}
 	rdr.expTableName = utils.CDRsTBL
-	if vals, has := processedOpt[utils.SQLTableName]; has {
+	if vals, has := processedOpt[utils.SQLTableNameOpt]; has {
 		rdr.expTableName = utils.IfaceAsString(vals)
 	}

--- a/ers/sql_test.go
+++ b/ers/sql_test.go
@@ -41,13 +41,13 @@ func TestSQLSetURL(t *testing.T) {
 	inURL := "*mysql://cgrates:CGRateS.org@127.0.0.1:3306"
 	outURL := "*mysql://cgrates:CGRateS.org@127.0.0.1:3306"
 	if err := sql.setURL(inURL, outURL, map[string]interface{}{
-		utils.SQLDBName:    "cgrates2",
-		utils.SQLTableName: "cdrs2",
-		"sslMode":          "enabled",
+		utils.SQLDBNameOpt:    "cgrates2",
+		utils.SQLTableNameOpt: "cdrs2",
+		utils.SSLModeCfg:      "enabled",

-		utils.SQLDBName + utils.ProcessedOpt:    "cgrates3",
-		utils.SQLTableName + utils.ProcessedOpt: "cdrs3",
-		"sslModeProcessed":                      "enabled",
+		utils.SQLDBNameOpt + utils.ProcessedOpt:    "cgrates3",
+		utils.SQLTableNameOpt + utils.ProcessedOpt: "cdrs3",
+		utils.SSLModeCfg + utils.ProcessedOpt:      "enabled",
 	}); err != nil {
 		t.Fatal(err)
 	} else if expsql.connString != sql.connString {
@@ -76,13 +76,13 @@ func TestSQLSetURL(t *testing.T) {
 	inURL = "*postgres://cgrates:CGRateS.org@127.0.0.1:3306"
 	outURL = "*postgres://cgrates:CGRateS.org@127.0.0.1:3306"
 	if err := sql.setURL(inURL, outURL, map[string]interface{}{
-		utils.SQLDBName:    "cgrates2",
-		utils.SQLTableName: "cdrs2",
-		"sslMode":          "enabled",
+		utils.SQLDBNameOpt:    "cgrates2",
+		utils.SQLTableNameOpt: "cdrs2",
+		utils.SSLModeCfg:      "enabled",

-		utils.SQLDBName + utils.ProcessedOpt:    "cgrates3",
-		utils.SQLTableName + utils.ProcessedOpt: "cdrs3",
-		"sslModeProcessed":                      "enabled",
+		utils.SQLDBNameOpt + utils.ProcessedOpt:    "cgrates3",
+		utils.SQLTableNameOpt + utils.ProcessedOpt: "cdrs3",
+		utils.SSLModeCfg + utils.ProcessedOpt:      "enabled",
 	}); err != nil {
 		t.Fatal(err)
 	} else if expsql.connString != sql.connString {
@@ -111,13 +111,13 @@ func TestSQLSetURL(t *testing.T) {
 	inURL = "*postgres://cgrates:CGRateS.org@127.0.0.1:3306"
 	outURL = ""
 	if err := sql.setURL(inURL, outURL, map[string]interface{}{
-		utils.SQLDBName:    "cgrates2",
-		utils.SQLTableName: "cdrs2",
-		"sslMode":          "enabled",
+		utils.SQLDBNameOpt:    "cgrates2",
+		utils.SQLTableNameOpt: "cdrs2",
+		utils.SSLModeCfg:      "enabled",

-		utils.SQLDBName + utils.ProcessedOpt:    "cgrates2",
-		utils.SQLTableName + utils.ProcessedOpt: "cdrs2",
-		"sslModeProcessed":                      "enabled",
+		utils.SQLDBNameOpt + utils.ProcessedOpt:    "cgrates2",
+		utils.SQLTableNameOpt + utils.ProcessedOpt: "cdrs2",
+		utils.SSLModeCfg + utils.ProcessedOpt:      "enabled",
 	}); err != nil {
 		t.Fatal(err)
 	} else if expsql.connString != sql.connString {
--- a/utils/consts.go
+++ b/utils/consts.go
@@ -2336,8 +2336,8 @@ const (
 	SQLDefaultDBName  = "cgrates"
 	SQLDefaultSSLMode = "disable"

-	SQLDBName    = "sqlDBName"
-	SQLTableName = "sqlTableName"
+	SQLDBNameOpt    = "sqlDBName"
+	SQLTableNameOpt = "sqlTableName"

 	// fileCSV
 	RowLengthOpt        = "RowLength"
@@ -2353,6 +2353,8 @@ const (
 	FlatstorePrfx            = "fst"
 	FstFailedCallsPrefixOpt  = "fstFailedCallsPrefix"
 	FstPartialRecordCacheOpt = "fstRecordCacheTTL"
+	FstMethodOpt             = "fstMethod" //
+	FstMadatoryACKOpt        = "fstMadatoryACK"

 	// fileXML
 	XMLRootPathOpt = "xmlRootPath"