Add test for internal storage for filter indexer

cmd/cgr-loader/cgr-loader.go

@@ -229,7 +229,6 @@ func main() {
 
 	if !*dryRun {
 		//tpid_remove
-
 		if *toStorDB { // Import files from a directory into storDb
 			if ldrCfg.LoaderCgrConfig.TpID == "" {
 				log.Fatal("TPid required.")
@@ -441,7 +440,6 @@ func main() {
 			}
 
 		}
-
 	} else {
 		if err := tpReader.RemoveFromDatabase(*verbose, *disableReverse); err != nil {
 			log.Fatal("Could not delete from database: ", err)

data/conf/samples/internal/cgrates.json

@@ -16,8 +16,8 @@
 
 
 "data_db": {
-	"db_type": "*internal",
-},
+    "db_type": "*internal",                   
+ },
 
 
 "stor_db": {

engine/cdr_it_test.go

@@ -43,7 +43,7 @@ func TestHttpJsonPost(t *testing.T) {
 		Usage: "0.00000001", ExtraFields: map[string]string{"field_extr1": "val_extr1", "fieldextr2": "valextr2"}, Cost: 1.01,
 	}
 	jsn, _ := json.Marshal(cdrOut)
-	if _, err := utils.HttpJsonPost("http://localhost:8000", false, jsn); err == nil {
+	if _, err := HttpJsonPost("http://localhost:8000", false, jsn); err == nil {
 		t.Error(err)
 	}
 }

engine/filterindexer.go

@@ -194,15 +194,5 @@ func (rfi *FilterIndexer) RemoveItemFromIndex(itemID string) (err error) {
 		}
 	}
 	rfi.reveseIndex[itemID] = make(utils.StringMap) //Force deleting in driver
-	if err = rfi.dm.SetFilterIndexes(
-		utils.PrefixToIndexCache[rfi.itemType], rfi.dbKeySuffix,
-		rfi.indexes, false, utils.NonTransactional); err != nil {
-		return
-	}
-	if err = rfi.dm.SetFilterReverseIndexes(
-		utils.PrefixToRevIndexCache[rfi.itemType], rfi.dbKeySuffix,
-		rfi.reveseIndex, false, utils.NonTransactional); err != nil {
-		return
-	}
-	return
+	return rfi.StoreIndexes(false, utils.NonTransactional)
 }

engine/filterindexer_it_test.go

@@ -93,6 +93,17 @@ func TestFilterIndexerITMongo(t *testing.T) {
 	}
 }
 
+func TestFilterIndexerITInternal(t *testing.T) {
+	mapDataDB, err := NewMapStorage()
+	if err != nil {
+		t.Fatal(err)
+	}
+	dataManager = NewDataManager(mapDataDB)
+	for _, stest := range sTests {
+		t.Run("TestITInternal", stest)
+	}
+}
+
 func testITFlush(t *testing.T) {
 	if err := dataManager.DataDB().Flush(""); err != nil {
 		t.Error(err)
@@ -265,7 +276,6 @@ func testITTestThresholdFilterIndexes(t *testing.T) {
 		Weight:             1.4,
 		ActionIDs:          []string{},
 	}
-
 	if err := dataManager.SetThresholdProfile(th, true); err != nil {
 		t.Error(err)
 	}

engine/poster_it_test.go

@@ -26,6 +26,8 @@ import (
 	"reflect"
 	"testing"
 	"time"
+
+	"github.com/cgrates/cgrates/utils"
 )
 
 type TestContent struct {
@@ -38,7 +40,7 @@ func TestHttpJsonPoster(t *testing.T) {
 	content := &TestContent{Var1: "Val1", Var2: "Val2"}
 	jsn, _ := json.Marshal(content)
 	filePath := "/tmp/cgr_test_http_poster.json"
-	if _, err := NewHTTPPoster(true, time.Duration(2*time.Second)).Post("http://localhost:8080/invalid", CONTENT_JSON, jsn, 3, filePath); err != nil {
+	if _, err := NewHTTPPoster(true, time.Duration(2*time.Second)).Post("http://localhost:8080/invalid", utils.CONTENT_JSON, jsn, 3, filePath); err != nil {
 		t.Error(err)
 	}
 	if readBytes, err := ioutil.ReadFile(filePath); err != nil {
@@ -57,7 +59,7 @@ func TestHttpBytesPoster(t *testing.T) {
 		Test2
 		`)
 	filePath := "/tmp/test_http_poster.http"
-	if _, err := NewHTTPPoster(true, time.Duration(2*time.Second)).Post("http://localhost:8080/invalid", CONTENT_TEXT, content, 3, filePath); err != nil {
+	if _, err := NewHTTPPoster(true, time.Duration(2*time.Second)).Post("http://localhost:8080/invalid", utils.CONTENT_TEXT, content, 3, filePath); err != nil {
 		t.Error(err)
 	}
 	if readBytes, err := ioutil.ReadFile(filePath); err != nil {

engine/storage_map_datadb.go

@@ -1251,7 +1251,9 @@ func (ms *MapStorage) GetFilterIndexesDrv(cacheID, itemIDPrefix, filterType stri
 				indexes[utils.ConcatenatedKey(filterType, fldName, fldVal)] = make(utils.StringMap)
 			}
 			if len(rcvidx[utils.ConcatenatedKey(filterType, fldName, fldVal)]) != 0 {
-				indexes[utils.ConcatenatedKey(filterType, fldName, fldVal)] = rcvidx[utils.ConcatenatedKey(fldName, fldVal)]
+				for key := range rcvidx[utils.ConcatenatedKey(filterType, fldName, fldVal)] {
+					indexes[utils.ConcatenatedKey(filterType, fldName, fldVal)][key] = true
+				}
 			}
 		}
 		return
@@ -1277,27 +1279,51 @@ func (ms *MapStorage) SetFilterIndexesDrv(cacheID, itemIDPrefix string,
 		dbKey = "tmp_" + utils.ConcatenatedKey(dbKey, transactionID)
 	}
 	if commit && transactionID != "" {
+		values, _ := ms.dict[dbKey]
 		delete(ms.dict, dbKey)
-		result, err := ms.ms.Marshal(indexes)
-		if err != nil {
-			return err
+		ms.dict[originKey] = values
+		return
+	}
+	var toBeDeleted []string
+	toBeAdded := make(map[string]utils.StringMap)
+	for key, strMp := range indexes {
+		if len(strMp) == 0 { // remove with no more elements inside
+			toBeDeleted = append(toBeDeleted, key)
+			delete(indexes, key)
+			continue
 		}
-		ms.dict[originKey] = result
-		return nil
-	} else {
-		for key, strMp := range indexes {
-			if len(strMp) == 0 { // remove with no more elements inside
-				delete(indexes, key)
-				continue
-			}
-		}
-		result, err := ms.ms.Marshal(indexes)
+		toBeAdded[key] = make(utils.StringMap)
+		toBeAdded[key] = strMp
+	}
+	values, has := ms.dict[dbKey]
+	if !has {
+		result, err := ms.ms.Marshal(toBeAdded)
 		if err != nil {
 			return err
 		}
 		ms.dict[dbKey] = result
-		return nil
+		return err
 	}
+	mp := make(map[string]utils.StringMap)
+	err = ms.ms.Unmarshal(values, &mp)
+	if err != nil {
+		return err
+	}
+	for _, key := range toBeDeleted {
+		delete(mp, key)
+	}
+	for key, strMp := range toBeAdded {
+		if _, has := mp[key]; !has {
+			mp[key] = make(utils.StringMap)
+		}
+		mp[key] = strMp
+	}
+	result, err := ms.ms.Marshal(mp)
+	if err != nil {
+		return err
+	}
+	ms.dict[dbKey] = result
+	return nil
 }
 
 func (ms *MapStorage) RemoveFilterIndexesDrv(cacheID, itemIDPrefix string) (err error) {
@@ -1324,11 +1350,13 @@ func (ms *MapStorage) GetFilterReverseIndexesDrv(cacheID, itemIDPrefix string,
 			return nil, err
 		}
 		indexes = make(map[string]utils.StringMap)
-		for fldName, fldVal := range fldNameVal {
-			if _, has := indexes[utils.ConcatenatedKey(fldName, fldVal)]; !has {
-				indexes[utils.ConcatenatedKey(fldName, fldVal)] = make(utils.StringMap)
+		for fldName, _ := range fldNameVal {
+			if _, has := indexes[fldName]; !has {
+				indexes[fldName] = make(utils.StringMap)
+			}
+			for key := range rcvidx[fldName] {
+				indexes[fldName][key] = true
 			}
-			indexes[utils.ConcatenatedKey(fldName, fldVal)] = rcvidx[utils.ConcatenatedKey(fldName, fldVal)]
 		}
 		return
 	} else {
@@ -1354,27 +1382,51 @@ func (ms *MapStorage) SetFilterReverseIndexesDrv(cacheID, itemIDPrefix string,
 		dbKey = "tmp_" + utils.ConcatenatedKey(dbKey, transactionID)
 	}
 	if commit && transactionID != "" {
+		values, _ := ms.dict[dbKey]
 		delete(ms.dict, dbKey)
-		result, err := ms.ms.Marshal(revIdx)
-		if err != nil {
-			return err
-		}
-		ms.dict[originKey] = result
+		ms.dict[originKey] = values
 		return nil
-	} else {
-		for key, strMp := range revIdx {
-			if len(strMp) == 0 { // remove with no more elements inside
-				delete(revIdx, key)
-				continue
-			}
+	}
+	var toBeDeleted []string
+	toBeAdded := make(map[string]utils.StringMap)
+	for key, strMp := range revIdx {
+		if len(strMp) == 0 { // remove with no more elements inside
+			toBeDeleted = append(toBeDeleted, key)
+			delete(revIdx, key)
+			continue
 		}
-		result, err := ms.ms.Marshal(revIdx)
+		toBeAdded[key] = make(utils.StringMap)
+		toBeAdded[key] = strMp
+	}
+	values, has := ms.dict[dbKey]
+	if !has {
+		result, err := ms.ms.Marshal(toBeAdded)
 		if err != nil {
 			return err
 		}
 		ms.dict[dbKey] = result
-		return nil
+		return err
 	}
+	mp := make(map[string]utils.StringMap)
+	err = ms.ms.Unmarshal(values, &mp)
+	if err != nil {
+		return err
+	}
+	for _, key := range toBeDeleted {
+		delete(mp, key)
+	}
+	for key, strMp := range toBeAdded {
+		if _, has := mp[key]; !has {
+			mp[key] = make(utils.StringMap)
+		}
+		mp[key] = strMp
+	}
+	result, err := ms.ms.Marshal(mp)
+	if err != nil {
+		return err
+	}
+	ms.dict[dbKey] = result
+	return
 }
 
 //RemoveFilterReverseIndexesDrv removes ReverseIndexes for a specific itemID

engine/storage_map_stordb.go

@@ -24,164 +24,164 @@ import (
 
 //implement LoadReader interface
 func (ms *MapStorage) GetTpIds(colName string) (ids []string, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTpTableIds(tpid, table string, distinct utils.TPDistinctIds,
 	filters map[string]string, paginator *utils.Paginator) (ids []string, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPTimings(tpid, id string) (timings []*utils.ApierTPTiming, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPDestinations(tpid, id string) (dsts []*utils.TPDestination, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPRates(tpid, id string) (rates []*utils.TPRate, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPDestinationRates(tpid, id string,
 	paginator *utils.Paginator) (dRates []*utils.TPDestinationRate, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPRatingPlans(string, string, *utils.Paginator) (rPlans []*utils.TPRatingPlan, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPRatingProfiles(filter *utils.TPRatingProfile) (rProfiles []*utils.TPRatingProfile, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPSharedGroups(tpid, id string) (sGroups []*utils.TPSharedGroups, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPCdrStats(tpid, id string) (stats []*utils.TPCdrStats, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPLCRs(filter *utils.TPLcrRules) (lcrs []*utils.TPLcrRules, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPUsers(filter *utils.TPUsers) (users []*utils.TPUsers, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPAliases(filter *utils.TPAliases) (aliases []*utils.TPAliases, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPDerivedChargers(*utils.TPDerivedChargers) (dCharges []*utils.TPDerivedChargers, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPActions(tpid, id string) (actions []*utils.TPActions, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPActionPlans(tpid, id string) (aPlans []*utils.TPActionPlan, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPActionTriggers(tpid, id string) (aTriggers []*utils.TPActionTriggers, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPAccountActions(filter *utils.TPAccountActions) (accounts []*utils.TPAccountActions, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPResources(tpid, id string) (resources []*utils.TPResource, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPStats(tpid, id string) (stats []*utils.TPStats, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPThresholds(tpid, id string) (ths []*utils.TPThreshold, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPFilters(tpid, id string) (fltrs []*utils.TPFilterProfile, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPSuppliers(tpid, id string) (supps []*utils.TPSupplierProfile, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetTPAttributes(tpid, id string) (attrs []*utils.TPAttributeProfile, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }
 
 //implement LoadWriter interface
 func (ms *MapStorage) RemTpData(table, tpid string, args map[string]string) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPTimings(timings []*utils.ApierTPTiming) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPDestinations(dests []*utils.TPDestination) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPRates(rates []*utils.TPRate) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPDestinationRates(dRates []*utils.TPDestinationRate) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPRatingPlans(ratingPlans []*utils.TPRatingPlan) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPRatingProfiles(ratingProfiles []*utils.TPRatingProfile) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPSharedGroups(groups []*utils.TPSharedGroups) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPCdrStats(cdrStats []*utils.TPCdrStats) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPUsers(users []*utils.TPUsers) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPAliases(aliases []*utils.TPAliases) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPDerivedChargers(dc []*utils.TPDerivedChargers) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPLCRs(lcrs []*utils.TPLcrRules) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPActions(acts []*utils.TPActions) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPActionPlans(aPlans []*utils.TPActionPlan) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPActionTriggers(aTriggers []*utils.TPActionTriggers) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPAccountActions(accActions []*utils.TPAccountActions) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPResources(resources []*utils.TPResource) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPStats(stats []*utils.TPStats) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 
 func (ms *MapStorage) SetTPThresholds(thresholds []*utils.TPThreshold) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 
 func (ms *MapStorage) SetTPFilters(filters []*utils.TPFilterProfile) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPSuppliers(suppliers []*utils.TPSupplierProfile) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) SetTPAttributes(attributes []*utils.TPAttributeProfile) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 
 //implement CdrStorage interface
 func (ms *MapStorage) SetCDR(cdr *CDR, allowUpdate bool) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) RemoveSMCost(smc *SMCost) (err error) {
-	return
+	return utils.ErrNotImplemented
 }
 func (ms *MapStorage) GetCDRs(filter *utils.CDRsFilter, remove bool) (cdrs []*CDR, count int64, err error) {
-	return
+	return nil, 0, utils.ErrNotImplemented
 }
 
 func (ms *MapStorage) GetSMCosts(cgrid, runid, originHost, originIDPrfx string) (smCosts []*SMCost, err error) {
-	return
+	return nil, utils.ErrNotImplemented
 }