engine.Responder with ProcessCdr method, moved cdrs and mediator to engine

engine/cdrs.go

@@ -0,0 +1,91 @@
+/*
+Rating system designed to be used in VoIP Carriers World
+Copyright (C) 2013 ITsysCOM
+
+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 engine
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/utils"
+)
+
+var (
+	cfg     *config.CGRConfig // Share the configuration with the rest of the package
+	storage CdrStorage
+	medi    *Mediator
+)
+
+// Returns error if not able to properly store the CDR, mediation is async since we can always recover offline
+func storeAndMediate(storedCdr *utils.StoredCdr) error {
+	if err := storage.SetCdr(storedCdr); err != nil {
+		return err
+	}
+	if cfg.CDRSMediator == utils.INTERNAL {
+		go func() {
+			if err := medi.RateCdr(storedCdr); err != nil {
+				Logger.Err(fmt.Sprintf("Could not run mediation on CDR: %s", err.Error()))
+			}
+		}()
+	}
+	return nil
+}
+
+// Handler for generic cgr cdr http
+func cgrCdrHandler(w http.ResponseWriter, r *http.Request) {
+	cgrCdr, err := utils.NewCgrCdrFromHttpReq(r)
+	if err != nil {
+		Logger.Err(fmt.Sprintf("Could not create CDR entry: %s", err.Error()))
+	}
+	if err := storeAndMediate(cgrCdr.AsStoredCdr()); err != nil {
+		Logger.Err(fmt.Sprintf("Errors when storing CDR entry: %s", err.Error()))
+	}
+}
+
+// Handler for fs http
+func fsCdrHandler(w http.ResponseWriter, r *http.Request) {
+	body, _ := ioutil.ReadAll(r.Body)
+	fsCdr, err := NewFSCdr(body)
+	if err != nil {
+		Logger.Err(fmt.Sprintf("Could not create CDR entry: %s", err.Error()))
+	}
+	if err := storeAndMediate(fsCdr.AsStoredCdr()); err != nil {
+		Logger.Err(fmt.Sprintf("Errors when storing CDR entry: %s", err.Error()))
+	}
+}
+
+type CDRS struct{}
+
+func NewCdrS(s CdrStorage, m *Mediator, c *config.CGRConfig) *CDRS {
+	storage = s
+	medi = m
+	cfg = c
+	return &CDRS{}
+}
+
+func (cdrs *CDRS) RegisterHanlersToServer(server *Server) {
+	server.RegisterHttpFunc("/cgr", cgrCdrHandler)
+	server.RegisterHttpFunc("/freeswitch_json", fsCdrHandler)
+}
+
+// Used to internally process CDR
+func (cdrs *CDRS) ProcessCdr(cdr *utils.StoredCdr) error {
+	return storeAndMediate(cdr)
+}

engine/fscdr.go

@@ -0,0 +1,143 @@
+/*
+Real-time Charging System for Telecom & ISP environments
+Copyright (C) 2012-2014 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 engine
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strings"
+
+	"github.com/cgrates/cgrates/utils"
+)
+
+const (
+	// Freswitch event property names
+	FS_CDR_MAP      = "variables"
+	FS_DIRECTION    = "direction"
+	FS_SUBJECT      = "cgr_subject"
+	FS_ACCOUNT      = "cgr_account"
+	FS_DESTINATION  = "cgr_destination"
+	FS_REQTYPE      = "cgr_reqtype" //prepaid or postpaid
+	FS_CATEGORY     = "cgr_category"
+	FS_UUID         = "uuid" // -Unique ID for this call leg
+	FS_CSTMID       = "cgr_tenant"
+	FS_CALL_DEST_NR = "dialed_extension"
+	FS_PARK_TIME    = "start_epoch"
+	FS_SETUP_TIME   = "start_epoch"
+	FS_ANSWER_TIME  = "answer_epoch"
+	FS_HANGUP_TIME  = "end_epoch"
+	FS_DURATION     = "billsec"
+	FS_USERNAME     = "user_name"
+	FS_IP           = "sip_local_network_addr"
+	FS_CDR_SOURCE   = "freeswitch_json"
+	FS_SIP_REQUSER  = "sip_req_user" // Apps like FusionPBX do not set dialed_extension, alternative being destination_number but that comes in customer profile, not in vars
+)
+
+func NewFSCdr(body []byte) (*FSCdr, error) {
+	fsCdr := new(FSCdr)
+	fsCdr.vars = make(map[string]string)
+	var err error
+	if err = json.Unmarshal(body, &fsCdr.body); err == nil {
+		if variables, ok := fsCdr.body[FS_CDR_MAP]; ok {
+			if variables, ok := variables.(map[string]interface{}); ok {
+				for k, v := range variables {
+					fsCdr.vars[k] = v.(string)
+				}
+			}
+			return fsCdr, nil
+		}
+	}
+	return nil, err
+}
+
+type FSCdr struct {
+	vars map[string]string
+	body map[string]interface{} // keeps the loaded body for extra field search
+}
+
+func (fsCdr FSCdr) getCgrId() string {
+	setupTime, _ := utils.ParseTimeDetectLayout(fsCdr.vars[FS_SETUP_TIME])
+	return utils.Sha1(fsCdr.vars[FS_UUID], setupTime.UTC().String())
+}
+
+func (fsCdr FSCdr) getExtraFields() map[string]string {
+	extraFields := make(map[string]string, len(cfg.CDRSExtraFields))
+	for _, field := range cfg.CDRSExtraFields {
+		origFieldVal, foundInVars := fsCdr.vars[field.Id]
+		if strings.HasPrefix(field.Id, utils.STATIC_VALUE_PREFIX) { // Support for static values injected in the CDRS. it will show up as {^value:value}
+			foundInVars = true
+		}
+		if !foundInVars {
+			origFieldVal = fsCdr.searchExtraField(field.Id, fsCdr.body)
+		}
+		extraFields[field.Id] = field.ParseValue(origFieldVal)
+	}
+	return extraFields
+}
+
+func (fsCdr FSCdr) searchExtraField(field string, body map[string]interface{}) (result string) {
+	for key, value := range body {
+		switch v := value.(type) {
+		case string:
+			if key == field {
+				return v
+			}
+		case map[string]interface{}:
+			if result = fsCdr.searchExtraField(field, v); result != "" {
+				return
+			}
+		case []interface{}:
+			for _, item := range v {
+				if otherMap, ok := item.(map[string]interface{}); ok {
+					if result = fsCdr.searchExtraField(field, otherMap); result != "" {
+						return
+					}
+				} else {
+					Logger.Warning(fmt.Sprintf("Slice with no maps: %v", reflect.TypeOf(item)))
+				}
+			}
+		default:
+			Logger.Warning(fmt.Sprintf("Unexpected type: %v", reflect.TypeOf(v)))
+		}
+	}
+	return
+}
+
+func (fsCdr FSCdr) AsStoredCdr() *utils.StoredCdr {
+	storCdr := new(utils.StoredCdr)
+	storCdr.CgrId = fsCdr.getCgrId()
+	storCdr.TOR = utils.VOICE
+	storCdr.AccId = fsCdr.vars[FS_UUID]
+	storCdr.CdrHost = fsCdr.vars[FS_IP]
+	storCdr.CdrSource = FS_CDR_SOURCE
+	storCdr.ReqType = utils.FirstNonEmpty(fsCdr.vars[FS_REQTYPE], cfg.DefaultReqType)
+	storCdr.Direction = "*out"
+	storCdr.Tenant = utils.FirstNonEmpty(fsCdr.vars[FS_CSTMID], cfg.DefaultTenant)
+	storCdr.Category = utils.FirstNonEmpty(fsCdr.vars[FS_CATEGORY], cfg.DefaultCategory)
+	storCdr.Account = utils.FirstNonEmpty(fsCdr.vars[FS_ACCOUNT], fsCdr.vars[FS_USERNAME])
+	storCdr.Subject = utils.FirstNonEmpty(fsCdr.vars[FS_SUBJECT], fsCdr.vars[FS_USERNAME])
+	storCdr.Destination = utils.FirstNonEmpty(fsCdr.vars[FS_DESTINATION], fsCdr.vars[FS_CALL_DEST_NR], fsCdr.vars[FS_SIP_REQUSER])
+	storCdr.SetupTime, _ = utils.ParseTimeDetectLayout(fsCdr.vars[FS_SETUP_TIME]) // Not interested to process errors, should do them if necessary in a previous step
+	storCdr.AnswerTime, _ = utils.ParseTimeDetectLayout(fsCdr.vars[FS_ANSWER_TIME])
+	storCdr.Usage, _ = utils.ParseDurationWithSecs(fsCdr.vars[FS_DURATION])
+	storCdr.ExtraFields = fsCdr.getExtraFields()
+	storCdr.Cost = -1
+	return storCdr
+}

engine/loader_local_test.go

@@ -44,7 +44,7 @@ README:
 var ratingDbCsv, ratingDbStor, ratingDbApier RatingStorage        // Each ratingDb will have it's own sources to collect data
 var accountDbCsv, accountDbStor, accountDbApier AccountingStorage // Each ratingDb will have it's own sources to collect data
 var storDb LoadStorage
-var cfg *config.CGRConfig
+var lCfg *config.CGRConfig
 
 // Arguments received via test command
 var testLocal = flag.Bool("local", false, "Perform the tests only on local test environment, not by default.") // This flag will be passed here via "go test -local" args
@@ -57,27 +57,27 @@ func TestConnDataDbs(t *testing.T) {
 	if !*testLocal {
 		return
 	}
-	cfg, _ = config.NewDefaultCGRConfig()
+	lCfg, _ = config.NewDefaultCGRConfig()
 	var err error
-	if ratingDbCsv, err = ConfigureRatingStorage(cfg.RatingDBType, cfg.RatingDBHost, cfg.RatingDBPort, "4", cfg.RatingDBUser, cfg.RatingDBPass, cfg.DBDataEncoding); err != nil {
+	if ratingDbCsv, err = ConfigureRatingStorage(lCfg.RatingDBType, lCfg.RatingDBHost, lCfg.RatingDBPort, "4", lCfg.RatingDBUser, lCfg.RatingDBPass, lCfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on ratingDb connection: ", err.Error())
 	}
-	if ratingDbStor, err = ConfigureRatingStorage(cfg.RatingDBType, cfg.RatingDBHost, cfg.RatingDBPort, "5", cfg.RatingDBUser, cfg.RatingDBPass, cfg.DBDataEncoding); err != nil {
+	if ratingDbStor, err = ConfigureRatingStorage(lCfg.RatingDBType, lCfg.RatingDBHost, lCfg.RatingDBPort, "5", lCfg.RatingDBUser, lCfg.RatingDBPass, lCfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on ratingDb connection: ", err.Error())
 	}
-	if ratingDbApier, err = ConfigureRatingStorage(cfg.RatingDBType, cfg.RatingDBHost, cfg.RatingDBPort, "6", cfg.RatingDBUser, cfg.RatingDBPass, cfg.DBDataEncoding); err != nil {
+	if ratingDbApier, err = ConfigureRatingStorage(lCfg.RatingDBType, lCfg.RatingDBHost, lCfg.RatingDBPort, "6", lCfg.RatingDBUser, lCfg.RatingDBPass, lCfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on ratingDb connection: ", err.Error())
 	}
-	if accountDbCsv, err = ConfigureAccountingStorage(cfg.AccountDBType, cfg.AccountDBHost, cfg.AccountDBPort, "7",
-		cfg.AccountDBUser, cfg.AccountDBPass, cfg.DBDataEncoding); err != nil {
+	if accountDbCsv, err = ConfigureAccountingStorage(lCfg.AccountDBType, lCfg.AccountDBHost, lCfg.AccountDBPort, "7",
+		lCfg.AccountDBUser, lCfg.AccountDBPass, lCfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on ratingDb connection: ", err.Error())
 	}
-	if accountDbStor, err = ConfigureAccountingStorage(cfg.AccountDBType, cfg.AccountDBHost, cfg.AccountDBPort, "8",
-		cfg.AccountDBUser, cfg.AccountDBPass, cfg.DBDataEncoding); err != nil {
+	if accountDbStor, err = ConfigureAccountingStorage(lCfg.AccountDBType, lCfg.AccountDBHost, lCfg.AccountDBPort, "8",
+		lCfg.AccountDBUser, lCfg.AccountDBPass, lCfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on ratingDb connection: ", err.Error())
 	}
-	if accountDbApier, err = ConfigureAccountingStorage(cfg.AccountDBType, cfg.AccountDBHost, cfg.AccountDBPort, "9",
-		cfg.AccountDBUser, cfg.AccountDBPass, cfg.DBDataEncoding); err != nil {
+	if accountDbApier, err = ConfigureAccountingStorage(lCfg.AccountDBType, lCfg.AccountDBHost, lCfg.AccountDBPort, "9",
+		lCfg.AccountDBUser, lCfg.AccountDBPass, lCfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on ratingDb connection: ", err.Error())
 	}
 	for _, db := range []Storage{ratingDbCsv, ratingDbStor, ratingDbApier, accountDbCsv, accountDbStor, accountDbApier} {
@@ -94,7 +94,7 @@ func TestCreateStorTpTables(t *testing.T) {
 		return
 	}
 	var db *MySQLStorage
-	if d, err := NewMySQLStorage(cfg.StorDBHost, cfg.StorDBPort, cfg.StorDBName, cfg.StorDBUser, cfg.StorDBPass); err != nil {
+	if d, err := NewMySQLStorage(lCfg.StorDBHost, lCfg.StorDBPort, lCfg.StorDBName, lCfg.StorDBUser, lCfg.StorDBPass); err != nil {
 		t.Error("Error on opening database connection: ", err)
 		return
 	} else {

engine/mediator.go

@@ -0,0 +1,178 @@
+/*
+Real-time Charging System for Telecom & ISP environments
+Copyright (C) 2012-2014 ITsysCOM GmbH
+
+This program is free software: you can Storagetribute 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 WITH*out 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 engine
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/utils"
+)
+
+func NewMediator(connector Connector, logDb LogStorage, cdrDb CdrStorage, cfg *config.CGRConfig) (m *Mediator, err error) {
+	m = &Mediator{
+		connector: connector,
+		logDb:     logDb,
+		cdrDb:     cdrDb,
+		cgrCfg:    cfg,
+	}
+	return m, nil
+}
+
+type Mediator struct {
+	connector Connector
+	logDb     LogStorage
+	cdrDb     CdrStorage
+	cgrCfg    *config.CGRConfig
+}
+
+// Retrive the cost from logging database, nil in case of no log
+func (self *Mediator) getCostsFromDB(cgrid, runId string) (cc *CallCost, err error) {
+	for i := 0; i < 3; i++ { // Mechanism to avoid concurrency between SessionManager writing the costs and mediator picking them up
+		cc, err = self.logDb.GetCallCostLog(cgrid, SESSION_MANAGER_SOURCE, runId)
+		if cc != nil {
+			break
+		}
+		time.Sleep(time.Duration(i) * time.Second)
+	}
+	return
+}
+
+// Retrive the cost from engine
+func (self *Mediator) getCostFromRater(storedCdr *utils.StoredCdr) (*CallCost, error) {
+	cc := &CallCost{}
+	var err error
+	if storedCdr.Usage == time.Duration(0) { // failed call,  returning empty callcost, no error
+		return cc, nil
+	}
+	cd := CallDescriptor{
+		TOR:           storedCdr.TOR,
+		Direction:     storedCdr.Direction,
+		Tenant:        storedCdr.Tenant,
+		Category:      storedCdr.Category,
+		Subject:       storedCdr.Subject,
+		Account:       storedCdr.Account,
+		Destination:   storedCdr.Destination,
+		TimeStart:     storedCdr.AnswerTime,
+		TimeEnd:       storedCdr.AnswerTime.Add(storedCdr.Usage),
+		DurationIndex: storedCdr.Usage,
+	}
+	if utils.IsSliceMember([]string{utils.PSEUDOPREPAID, utils.POSTPAID}, storedCdr.ReqType) {
+		err = self.connector.Debit(cd, cc)
+	} else {
+		err = self.connector.GetCost(cd, cc)
+	}
+	if err != nil {
+		self.logDb.LogError(storedCdr.CgrId, MEDIATOR_SOURCE, storedCdr.MediationRunId, err.Error())
+	} else {
+		// If the mediator calculated a price it will write it to logdb
+		self.logDb.LogCallCost(storedCdr.CgrId, MEDIATOR_SOURCE, storedCdr.MediationRunId, cc)
+	}
+	return cc, err
+}
+
+func (self *Mediator) rateCDR(storedCdr *utils.StoredCdr) error {
+	var qryCC *CallCost
+	var errCost error
+	if storedCdr.ReqType == utils.PREPAID {
+		// Should be previously calculated and stored in DB
+		qryCC, errCost = self.getCostsFromDB(storedCdr.CgrId, storedCdr.MediationRunId)
+	} else {
+		qryCC, errCost = self.getCostFromRater(storedCdr)
+	}
+	if errCost != nil {
+		return errCost
+	} else if qryCC == nil {
+		return errors.New("No cost returned from rater")
+	}
+	storedCdr.Cost = qryCC.Cost
+	return nil
+}
+
+func (self *Mediator) RateCdr(storedCdr *utils.StoredCdr) error {
+	storedCdr.MediationRunId = utils.DEFAULT_RUNID
+	cdrRuns := []*utils.StoredCdr{storedCdr}     // Start with initial storCdr, will add here all to be mediated
+	attrsDC := utils.AttrDerivedChargers{Tenant: storedCdr.Tenant, Category: storedCdr.Category, Direction: storedCdr.Direction,
+		Account: storedCdr.Account, Subject: storedCdr.Subject}
+	var dcs utils.DerivedChargers
+	if err := self.connector.GetDerivedChargers(attrsDC, &dcs); err != nil {
+		errText := fmt.Sprintf("Could not get derived charging for cgrid %s, error: %s", storedCdr.CgrId, err.Error())
+		Logger.Err(errText)
+		return errors.New(errText)
+	}
+	for _, dc := range dcs {
+		runFilters, _ := utils.ParseRSRFields(dc.RunFilters, utils.INFIELD_SEP)
+		matchingAllFilters := true
+		for _, dcRunFilter := range runFilters {
+			if fltrPass, _ := storedCdr.PassesFieldFilter(dcRunFilter); !fltrPass {
+				matchingAllFilters = false
+				break
+			}
+		}
+		if !matchingAllFilters { // Do not process the derived charger further if not all filters were matched
+			continue
+		}
+		dcReqTypeFld, _ := utils.NewRSRField(dc.ReqTypeField)
+		dcDirFld, _ := utils.NewRSRField(dc.DirectionField)
+		dcTenantFld, _ := utils.NewRSRField(dc.TenantField)
+		dcCategoryFld, _ := utils.NewRSRField(dc.CategoryField)
+		dcAcntFld, _ := utils.NewRSRField(dc.AccountField)
+		dcSubjFld, _ := utils.NewRSRField(dc.SubjectField)
+		dcDstFld, _ := utils.NewRSRField(dc.DestinationField)
+		dcSTimeFld, _ := utils.NewRSRField(dc.SetupTimeField)
+		dcATimeFld, _ := utils.NewRSRField(dc.AnswerTimeField)
+		dcDurFld, _ := utils.NewRSRField(dc.UsageField)
+		forkedCdr, err := storedCdr.ForkCdr(dc.RunId, dcReqTypeFld, dcDirFld, dcTenantFld, dcCategoryFld, dcAcntFld, dcSubjFld, dcDstFld, dcSTimeFld, dcATimeFld, dcDurFld,
+			[]*utils.RSRField{}, true)
+		if err != nil { // Errors on fork, cannot calculate further, write that into db for later analysis
+			self.cdrDb.SetRatedCdr(&utils.StoredCdr{CgrId: storedCdr.CgrId, CdrSource: utils.FORKED_CDR, MediationRunId: dc.RunId, Cost: -1},
+				err.Error()) // Cannot fork CDR, important just runid and error
+			continue
+		}
+		cdrRuns = append(cdrRuns, forkedCdr)
+	}
+	for _, cdr := range cdrRuns {
+		extraInfo := ""
+		if err := self.rateCDR(cdr); err != nil {
+			extraInfo = err.Error()
+		}
+		if err := self.cdrDb.SetRatedCdr(cdr, extraInfo); err != nil {
+			Logger.Err(fmt.Sprintf("<Mediator> Could not record cost for cgrid: <%s>, ERROR: <%s>, cost: %f, extraInfo: %s",
+				cdr.CgrId, err.Error(), cdr.Cost, extraInfo))
+		}
+	}
+	return nil
+}
+
+func (self *Mediator) RateCdrs(cgrIds, runIds, tors, cdrHosts, cdrSources, reqTypes, directions, tenants, categories, accounts, subjects, destPrefixes, ratedAccounts, ratedSubjects []string,
+	orderIdStart, orderIdEnd int64, timeStart, timeEnd time.Time, rerateErrors, rerateRated bool) error {
+	cdrs, err := self.cdrDb.GetStoredCdrs(cgrIds, runIds, tors, cdrHosts, cdrSources, reqTypes, directions, tenants, categories, accounts, subjects, destPrefixes, ratedAccounts, ratedSubjects,
+		orderIdStart, orderIdEnd, timeStart, timeEnd, !rerateErrors, !rerateRated, true)
+	if err != nil {
+		return err
+	}
+	for _, cdr := range cdrs {
+		if err := self.RateCdr(cdr); err != nil {
+			return err
+		}
+	}
+	return nil
+}

engine/mediator_local_test.go

@@ -0,0 +1,286 @@
+/*
+Rating system designed to be used in VoIP Carriers World
+Copyright (C) 2013 ITsysCOM
+
+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 engine
+
+import (
+	"flag"
+	"fmt"
+	"net/http"
+	"net/rpc"
+	"net/rpc/jsonrpc"
+	"net/url"
+	"os/exec"
+	"path"
+	"testing"
+	"time"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/utils"
+)
+
+/*
+README:
+
+ Enable local tests by passing '-local' to the go test command
+ It is expected that the data folder of CGRateS exists at path /usr/share/cgrates/data or passed via command arguments.
+ Prior running the tests, create database and users by running:
+  mysql -pyourrootpwd < /usr/share/cgrates/data/storage/mysql/create_db_with_users.sql
+ What these tests do:
+  * Flush tables in storDb to start clean.
+  * Start engine with default configuration and give it some time to listen (here caching can slow down, hence the command argument parameter).
+  * Connect rpc client depending on encoding defined in configuration.
+  * Execute remote Apis and test their replies(follow prepaid1cent scenario so we can test load in dataDb also).
+*/
+
+var cgrCfg *config.CGRConfig
+var cgrRpc *rpc.Client
+var cdrStor CdrStorage
+var httpClient *http.Client
+
+var storDbType = flag.String("stordb_type", utils.MYSQL, "The type of the storDb database <mysql>")
+var startDelay = flag.Int("delay_start", 300, "Number of miliseconds to it for rater to start and cache")
+var cfgPath = path.Join(*dataDir, "conf", "samples", "mediator_test1.cfg")
+
+func TestInitRatingDb(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	var err error
+	cgrCfg, err = config.NewCGRConfigFromFile(&cfgPath)
+	if err != nil {
+		t.Fatal("Got config error: ", err.Error())
+	}
+	ratingDb, err := ConfigureRatingStorage(cgrCfg.RatingDBType, cgrCfg.RatingDBHost, cgrCfg.RatingDBPort, cgrCfg.RatingDBName, cgrCfg.RatingDBUser, cgrCfg.RatingDBPass, cgrCfg.DBDataEncoding)
+	if err != nil {
+		t.Fatal("Cannot connect to dataDb", err)
+	}
+	if err := ratingDb.Flush(); err != nil {
+		t.Fatal("Cannot reset dataDb", err)
+	}
+}
+
+// Empty tables before using them
+func TestInitStorDb(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	if *storDbType != utils.MYSQL {
+		t.Fatal("Unsupported storDbType")
+	}
+	var mysql *MySQLStorage
+	var err error
+	if cdrStor, err = ConfigureCdrStorage(cgrCfg.StorDBType, cgrCfg.StorDBHost, cgrCfg.StorDBPort, cgrCfg.StorDBName, cgrCfg.StorDBUser, cgrCfg.StorDBPass); err != nil {
+		t.Fatal("Error on opening database connection: ", err)
+	} else {
+		mysql = cdrStor.(*MySQLStorage)
+	}
+	if err := mysql.CreateTablesFromScript(path.Join(*dataDir, "storage", *storDbType, CREATE_CDRS_TABLES_SQL)); err != nil {
+		t.Fatal("Error on mysql creation: ", err.Error())
+		return // No point in going further
+	}
+	for _, tbl := range []string{utils.TBL_CDRS_PRIMARY, utils.TBL_CDRS_EXTRA} {
+		if _, err := mysql.Db.Query(fmt.Sprintf("SELECT 1 from %s", tbl)); err != nil {
+			t.Fatal(err.Error())
+		}
+	}
+}
+
+// Finds cgr-engine executable and starts it with default configuration
+func TestStartEngine(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	enginePath, err := exec.LookPath("cgr-engine")
+	if err != nil {
+		t.Fatal("Cannot find cgr-engine executable")
+	}
+	exec.Command("pkill", "cgr-engine").Run() // Just to make sure another one is not running, bit brutal maybe we can fine tune it
+	engine := exec.Command(enginePath, "-config", cfgPath)
+	if err := engine.Start(); err != nil {
+		t.Fatal("Cannot start cgr-engine: ", err.Error())
+	}
+	time.Sleep(time.Duration(*startDelay) * time.Millisecond) // Give time to rater to fire up
+	httpClient = new(http.Client)
+}
+
+// Connect rpc client
+func TestRpcConn(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	var err error
+	//cgrRpc, err = rpc.Dial("tcp", cfg.RPCGOBListen) //ToDo: Fix with automatic config
+	cgrRpc, err = jsonrpc.Dial("tcp", cgrCfg.RPCJSONListen)
+	if err != nil {
+		t.Fatal("Could not connect to CGR GOB-RPC Server: ", err.Error())
+	}
+}
+
+func TestPostCdrs(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	cdrForm1 := url.Values{utils.TOR: []string{utils.VOICE}, utils.ACCID: []string{"dsafdsaf"}, utils.CDRHOST: []string{"192.168.1.1"}, utils.REQTYPE: []string{"rated"}, utils.DIRECTION: []string{"*out"},
+		utils.TENANT: []string{"cgrates.org"}, utils.CATEGORY: []string{"call"}, utils.ACCOUNT: []string{"2001"}, utils.SUBJECT: []string{"2001"},
+		utils.DESTINATION: []string{"+4986517174963"},
+		utils.ANSWER_TIME: []string{"2013-11-07T08:42:26Z"}, utils.USAGE: []string{"10"}, "field_extr1": []string{"val_extr1"}, "fieldextr2": []string{"valextr2"}}
+	cdrForm2 := url.Values{utils.TOR: []string{utils.VOICE}, utils.ACCID: []string{"adsafdsaf"}, utils.CDRHOST: []string{"192.168.1.1"}, utils.REQTYPE: []string{"rated"}, utils.DIRECTION: []string{"*out"},
+		utils.TENANT: []string{"itsyscom.com"}, utils.CATEGORY: []string{"call"}, utils.ACCOUNT: []string{"1003"}, utils.SUBJECT: []string{"1003"}, utils.DESTINATION: []string{"+4986517174964"},
+		utils.ANSWER_TIME: []string{"2013-11-07T08:42:26Z"}, utils.USAGE: []string{"10"}, "field_extr1": []string{"val_extr1"}, "fieldextr2": []string{"valextr2"}}
+	cdrFormData1 := url.Values{utils.TOR: []string{utils.DATA}, utils.ACCID: []string{"616350843"}, utils.CDRHOST: []string{"192.168.1.1"}, utils.REQTYPE: []string{"rated"},
+		utils.DIRECTION: []string{"*out"}, utils.TENANT: []string{"cgrates.org"}, utils.CATEGORY: []string{"data"},
+		utils.ACCOUNT: []string{"1010"}, utils.SUBJECT: []string{"1010"}, utils.ANSWER_TIME: []string{"2013-11-07T08:42:26Z"},
+		utils.USAGE: []string{"10"}, "field_extr1": []string{"val_extr1"}, "fieldextr2": []string{"valextr2"}}
+	for _, cdrForm := range []url.Values{cdrForm1, cdrForm2, cdrFormData1} {
+		cdrForm.Set(utils.CDRSOURCE, TEST_SQL)
+		if _, err := httpClient.PostForm(fmt.Sprintf("http://%s/cgr", cgrCfg.HTTPListen), cdrForm); err != nil {
+			t.Error(err.Error())
+		}
+	}
+	time.Sleep(100 * time.Millisecond) // Give time for CDRs to reach database
+	if storedCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, false, false, false); err != nil {
+		t.Error(err)
+	} else if len(storedCdrs) != 6 { // Make sure CDRs made it into StorDb
+		t.Error(fmt.Sprintf("Unexpected number of CDRs stored: %d", len(storedCdrs)))
+	}
+	if nonErrorCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, true, false, false); err != nil {
+		t.Error(err)
+	} else if len(nonErrorCdrs) != 0 {
+		t.Error(fmt.Sprintf("Unexpected number of CDRs stored: %d", len(nonErrorCdrs)))
+	}
+}
+
+// Directly inject CDRs into storDb
+func TestInjectCdrs(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	cgrCdr1 := utils.CgrCdr{utils.TOR: utils.VOICE, utils.ACCID: "aaaaadsafdsaf", "cdrsource": TEST_SQL, utils.CDRHOST: "192.168.1.1", utils.REQTYPE: "rated", utils.DIRECTION: "*out",
+		utils.TENANT: "cgrates.org", utils.CATEGORY: "call", utils.ACCOUNT: "dan", utils.SUBJECT: "dan", utils.DESTINATION: "+4986517174963",
+		utils.ANSWER_TIME: "2013-11-07T08:42:26Z", utils.USAGE: "10"}
+	cgrCdr2 := utils.CgrCdr{utils.TOR: utils.VOICE, utils.ACCID: "baaaadsafdsaf", "cdrsource": TEST_SQL, utils.CDRHOST: "192.168.1.1", utils.REQTYPE: "rated", utils.DIRECTION: "*out",
+		utils.TENANT: "cgrates.org", utils.CATEGORY: "call", utils.ACCOUNT: "dan", utils.SUBJECT: "dan", utils.DESTINATION: "+4986517173964",
+		utils.ANSWER_TIME: "2013-11-07T09:42:26Z", utils.USAGE: "20"}
+	for _, cdr := range []utils.CgrCdr{cgrCdr1, cgrCdr2} {
+		if err := cdrStor.SetCdr(cdr.AsStoredCdr()); err != nil {
+			t.Error(err)
+		}
+	}
+	if storedCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, false, false, false); err != nil {
+		t.Error(err)
+	} else if len(storedCdrs) != 8 { // Make sure CDRs made it into StorDb
+		t.Error(fmt.Sprintf("Unexpected number of CDRs stored: %d", len(storedCdrs)))
+	}
+	if nonRatedCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, true, true, false); err != nil {
+		t.Error(err)
+	} else if len(nonRatedCdrs) != 2 { // Just two of them should be non-rated
+		t.Error(fmt.Sprintf("Unexpected number of CDRs non-rated: %d", len(nonRatedCdrs)))
+	}
+}
+
+// Test here LoadTariffPlanFromFolder
+func TestLoadTariffPlanFromFolder(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	reply := ""
+	// Simple test that command is executed without errors
+	attrs := &utils.AttrLoadTpFromFolder{FolderPath: path.Join(*dataDir, "tariffplans", "prepaid1centpsec")}
+	if err := cgrRpc.Call("ApierV1.LoadTariffPlanFromFolder", attrs, &reply); err != nil {
+		t.Error("Got error on ApierV1.LoadTariffPlanFromFolder: ", err.Error())
+	} else if reply != utils.OK {
+		t.Error("Calling ApierV1.LoadTariffPlanFromFolder got reply: ", reply)
+	}
+}
+
+func TestRateCdrs(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	var reply string
+	if err := cgrRpc.Call("MediatorV1.RateCdrs", utils.AttrRateCdrs{}, &reply); err != nil {
+		t.Error(err.Error())
+	} else if reply != utils.OK {
+		t.Errorf("Unexpected reply: %s", reply)
+	}
+	if nonRatedCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, true, true, false); err != nil {
+		t.Error(err)
+	} else if len(nonRatedCdrs) != 0 { // All CDRs should be rated
+		t.Error(fmt.Sprintf("Unexpected number of CDRs non-rated: %d", len(nonRatedCdrs)))
+	}
+	if errRatedCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, false, true, false); err != nil {
+		t.Error(err)
+	} else if len(errRatedCdrs) != 8 { // The first 2 with errors should be still there before rerating
+		t.Error(fmt.Sprintf("Unexpected number of CDRs with errors: %d", len(errRatedCdrs)))
+	}
+	if err := cgrRpc.Call("MediatorV1.RateCdrs", utils.AttrRateCdrs{RerateErrors: true}, &reply); err != nil {
+		t.Error(err.Error())
+	} else if reply != utils.OK {
+		t.Errorf("Unexpected reply: %s", reply)
+	}
+	if errRatedCdrs, err := cdrStor.GetStoredCdrs(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, 0, 0, time.Time{}, time.Time{}, false, true, false); err != nil {
+		t.Error(err)
+	} else if len(errRatedCdrs) != 4 {
+		t.Error(fmt.Sprintf("Unexpected number of CDRs with errors: %d", len(errRatedCdrs)))
+	}
+}
+
+/*
+func TestMediatePseudoprepaid(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	var reply *engine.Account
+	attrs := &utils.AttrGetAccount{Tenant: "cgrates.org", Account: "1003", Direction: "*out"}
+	if err := cgrRpc.Call("ApierV1.GetAccount", attrs, &reply); err != nil {
+		t.Error("Got error on ApierV1.GetAccount: ", err.Error())
+	} else if reply.BalanceMap[engine.CREDIT+attrs.Direction].GetTotalValue() != 11 {
+		t.Errorf("Calling ApierV1.GetBalance expected: 10.0, received: %f", reply.BalanceMap[engine.CREDIT+attrs.Direction].GetTotalValue())
+	}
+	voiceCdr := &utils.StoredCdr{TOR: utils.VOICE, AccId: "dsafdsaf", CdrHost: "192.168.1.1", CdrSource: "test", ReqType: utils.PSEUDOPREPAID, Direction: utils.OUT,
+		Tenant: "cgrates.org", Category: "call", Account: "1003", Subject: "1003", Destination: "+4986517174963",
+		SetupTime: time.Date(2013, 11, 7, 8, 42, 26, 0, time.UTC), AnswerTime: time.Date(2013, 11, 7, 8, 42, 26, 0, time.UTC),
+		Usage: time.Duration(5) * time.Second}
+	dataCdr := &utils.StoredCdr{TOR: utils.DATA, AccId: "6163508432", CdrHost: "192.168.1.1", CdrSource: "test", ReqType: utils.PSEUDOPREPAID, Direction: utils.OUT,
+		Tenant: "cgrates.org", Category: "data", Account: "1003", Subject: "1003",
+		SetupTime: time.Date(2013, 11, 7, 8, 42, 26, 0, time.UTC), AnswerTime: time.Date(2013, 11, 7, 8, 42, 26, 0, time.UTC),
+		Usage: time.Duration(10) * time.Second}
+	for _, cdrForm := range []url.Values{voiceCdr.AsHttpForm(), dataCdr.AsHttpForm()} {
+		cdrForm.Set(utils.CDRSOURCE, engine.TEST_SQL)
+		if _, err := httpClient.PostForm(fmt.Sprintf("http://%s/cgr", cfg.HTTPListen), cdrForm); err != nil {
+			t.Error(err.Error())
+		}
+	}
+	time.Sleep(time.Duration(*startDelay) * time.Millisecond) // Give time for debits to happen
+	expectBalance := 5.998
+	if err := cgrRpc.Call("ApierV1.GetAccount", attrs, &reply); err != nil {
+		t.Error("Got error on ApierV1.GetAccount: ", err.Error())
+	} else if reply.BalanceMap[engine.CREDIT+attrs.Direction].GetTotalValue() != expectBalance { // 5 from voice, 0.002 from DATA
+		t.Errorf("Calling ApierV1.GetBalance expected: %f, received: %f", expectBalance, reply.BalanceMap[engine.CREDIT+attrs.Direction].GetTotalValue())
+	}
+}
+*/
+
+// Simply kill the engine after we are done with tests within this file
+func TestStopEngine(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	exec.Command("pkill", "cgr-engine").Run()
+}

engine/responder.go

@@ -36,6 +36,7 @@ import (
 type Responder struct {
 	Bal      *balancer2go.Balancer
 	ExitChan chan bool
+	CdrSrv   *CDRS
 }
 
 /*
@@ -128,6 +129,17 @@ func (rs *Responder) GetDerivedChargers(attrs utils.AttrDerivedChargers, dcs *ut
 	return nil
 }
 
+func (rs *Responder) ProcessCdr(cdr *utils.StoredCdr, reply *string) error {
+	if rs.CdrSrv == nil {
+		return errors.New("CdrServerNotRunning")
+	}
+	if err := rs.CdrSrv.ProcessCdr(cdr); err != nil {
+		return err
+	}
+	*reply = utils.OK
+	return nil
+}
+
 func (rs *Responder) FlushCache(arg CallDescriptor, reply *float64) (err error) {
 	if rs.Bal != nil {
 		*reply, err = rs.callMethod(&arg, "Responder.FlushCache")
@@ -287,6 +299,7 @@ type Connector interface {
 	RefundIncrements(CallDescriptor, *float64) error
 	GetMaxSessionTime(CallDescriptor, *float64) error
 	GetDerivedChargers(utils.AttrDerivedChargers, *utils.DerivedChargers) error
+	ProcessCdr(*utils.StoredCdr, *string) error
 }
 
 type RPCClientConnector struct {
@@ -316,3 +329,7 @@ func (rcc *RPCClientConnector) GetMaxSessionTime(cd CallDescriptor, resp *float6
 func (rcc *RPCClientConnector) GetDerivedChargers(attrs utils.AttrDerivedChargers, dcs *utils.DerivedChargers) error {
 	return rcc.Client.Call("ApierV1.GetDerivedChargers", attrs, dcs)
 }
+
+func (rcc *RPCClientConnector) ProcessCdr(cdr *utils.StoredCdr, reply *string) error {
+	return rcc.Client.Call("CDRSV1.ProcessCdr", cdr, reply)
+}