Skel of kamailio session manager

cmd/cgr-engine/cgr-engine.go

@@ -50,6 +50,7 @@ const (
 	REDIS    = "redis"
 	SAME     = "same"
 	FS       = "freeswitch"
+	KAMAILIO = "kamailio"
 	OSIPS    = "opensips"
 )
 
@@ -184,6 +185,8 @@ func startSessionManager(responder *engine.Responder, loggerDb engine.LogStorage
 	case FS:
 		dp, _ := time.ParseDuration(fmt.Sprintf("%vs", cfg.SMDebitInterval))
 		sm = sessionmanager.NewFSSessionManager(cfg, loggerDb, raterConn, cdrsConn, dp)
+	case KAMAILIO:
+		sm, _ = sessionmanager.NewKamailioSessionManager(cfg, raterConn, cdrsConn)
 	case OSIPS:
 		sm, _ = sessionmanager.NewOSipsSessionManager(cfg, raterConn, cdrsConn)
 	default:

config/config.go

@@ -127,6 +127,8 @@ type CGRConfig struct {
 	OsipsMiAddr             string                // Adress where to reach OpenSIPS mi_datagram module
 	OsipsEvSubscInterval    time.Duration         // Refresh event subscription at this interval
 	OsipsReconnects         int                   // Number of attempts on connect failure.
+	KamailioEvApiAddr       string                // Address of the kamailio evapi server
+	KamailioReconnects      int                   // Number of reconnect attempts on connection lost
 	HistoryAgentEnabled     bool                  // Starts History as an agent: <true|false>.
 	HistoryServer           string                // Address where to reach the master history server: <internal|x.y.z.y:1234>
 	HistoryServerEnabled    bool                  // Starts History as server: <true|false>.
@@ -212,6 +214,8 @@ func (self *CGRConfig) setDefaults() error {
 	self.OsipsMiAddr = "127.0.0.1:8020"
 	self.OsipsEvSubscInterval = time.Duration(60) * time.Second
 	self.OsipsReconnects = 3
+	self.KamailioEvApiAddr = "127.0.0.1:8448"
+	self.KamailioReconnects = 3
 	self.HistoryAgentEnabled = false
 	self.HistoryServerEnabled = false
 	self.HistoryServer = utils.INTERNAL
@@ -590,6 +594,12 @@ func loadConfig(c *conf.ConfigFile) (*CGRConfig, error) {
 	if hasOpt = c.HasOption("opensips", "reconnects"); hasOpt {
 		cfg.OsipsReconnects, _ = c.GetInt("opensips", "reconnects")
 	}
+	if hasOpt = c.HasOption("kamailio", "evapi_addr"); hasOpt {
+		cfg.KamailioEvApiAddr, _ = c.GetString("kamailio", "evapi_addr")
+	}
+	if hasOpt = c.HasOption("kamailio", "reconnects"); hasOpt {
+		cfg.KamailioReconnects, _ = c.GetInt("kamailio", "reconnects")
+	}
 	if cfg.DerivedChargers, err = ParseCfgDerivedCharging(c); err != nil {
 		return nil, err
 	}

config/config_test.go

@@ -116,6 +116,8 @@ func TestDefaults(t *testing.T) {
 	eCfg.OsipsMiAddr = "127.0.0.1:8020"
 	eCfg.OsipsEvSubscInterval = time.Duration(60) * time.Second
 	eCfg.OsipsReconnects = 3
+	eCfg.KamailioEvApiAddr = "127.0.0.1:8448"
+	eCfg.KamailioReconnects = 3
 	eCfg.DerivedChargers = make(utils.DerivedChargers, 0)
 	eCfg.CombinedDerivedChargers = true
 	eCfg.HistoryAgentEnabled = false
@@ -281,6 +283,8 @@ func TestConfigFromFile(t *testing.T) {
 	eCfg.OsipsMiAddr = "test"
 	eCfg.OsipsEvSubscInterval = time.Duration(99) * time.Second
 	eCfg.OsipsReconnects = 99
+	eCfg.KamailioEvApiAddr = "test"
+	eCfg.KamailioReconnects = 99
 	eCfg.DerivedChargers = utils.DerivedChargers{&utils.DerivedCharger{RunId: "test", RunFilters: "", ReqTypeField: "test", DirectionField: "test", TenantField: "test",
 		CategoryField: "test", AccountField: "test", SubjectField: "test", DestinationField: "test", SetupTimeField: "test", AnswerTimeField: "test", UsageField: "test"}}
 	eCfg.CombinedDerivedChargers = true

config/test_data.txt

@@ -135,6 +135,10 @@ empty_balance_context = test    # If defined, call will be transfered to this co
 empty_balance_ann_file = test  	# File to be played before disconnecting prepaid calls (applies only if no context defined)
 cdr_extra_fields = test			# Extra fields to store in CDRs in case of processing them
 
+[kamailio]
+evapi_addr = test
+reconnects = 99					# Number of attempts on connect failure.
+
 [opensips]
 listen_udp = test 				# Address where to listen for event datagrams coming from OpenSIPS
 mi_addr = test					# Adress where to reach OpenSIPS mi_datagram module

data/conf/cgrates.cfg

@@ -140,6 +140,10 @@
 # empty_balance_ann_file =		# File to be played before disconnecting prepaid calls on empty balance (applies only if no context defined)
 # cdr_extra_fields = 			# Extra fields to store in CDRs in case of processing them
 
+[kamailio]
+# evapi_addr = 127.0.0.1:8448 	# Address of the kamailio evapi server
+# reconnects = 3				# Number of attempts on connect failure.
+
 [opensips]
 # listen_udp = 127.0.0.1:2020 		# Address where to listen for datagram events coming from OpenSIPS
 # mi_addr = 127.0.0.1:8020			# Adress where to reach OpenSIPS mi_datagram module

data/conf/samples/kamailio_evapi.cfg

@@ -0,0 +1,60 @@
+# Real-time Charging System for Telecom & ISP environments
+# Copyright (C) ITsysCOM GmbH
+#
+# This file contains the default configuration hardcoded into CGRateS.
+# This is what you get when you load CGRateS with an empty configuration file.
+
+[global]
+rpc_json_listen = :2012	# RPC JSON listening address
+
+[rater]
+enabled = true				# Enable RaterCDRSExportPath service: <true|false>.
+
+[scheduler]
+enabled = true				# Starts Scheduler service: <true|false>.
+
+[cdrs]
+enabled = true					# Start the CDR Server service:  <true|false>.
+mediator = 	internal			# Address where to reach the Mediator. Empty for disabling mediation. <""|internal>
+# cdrstats =						# Address where to reach the cdrstats service: <internal|x.y.z.y:1234>
+
+[mediator]
+enabled = true				# Starts Mediator service: <true|false>.
+# rater = internal			# Address where to reach the Rater: <internal|x.y.z.y:1234>
+# cdrstats =	internal		# Address where to reach the cdrstats service: <internal|x.y.z.y:1234>
+
+[cdrstats]
+enabled = true						# Starts the cdrstats service: <true|false>
+#queue_length = 50					# Number of items in the stats buffer
+time_window = 1h					# Will only keep the CDRs who's call setup time is not older than time.Now()-TimeWindow
+# metrics = ASR, ACD, ACC			# Stat metric ids to build
+# setup_interval =					# Filter on CDR SetupTime
+# tors = 							# Filter on CDR TOR fields
+# cdr_hosts=						# Filter on CDR CdrHost fields
+# cdr_sources = 					# Filter on CDR CdrSource fields
+# req_types = 						# Filter on CDR ReqType fields
+# directions =						# Filter on CDR Direction fields
+# tenants = 						# Filter on CDR Tenant fields
+# categories = 						# Filter on CDR	Category fields
+# accounts =						# Filter on CDR Account fields
+# subjects = 						# Filter on CDR Subject fields
+# destination_prefixes =			# Filter on CDR Destination prefixes
+# usage_interval = 					# Filter on CDR Usage 
+# mediation_run_ids =				# Filter on CDR MediationRunId fields
+# rated_accounts =					# Filter on CDR RatedAccount fields
+# rated_subjects = 					# Filter on CDR RatedSubject fields
+# cost_intervals =					# Filter on CDR Cost
+
+[session_manager]
+enabled = true				# Starts SessionManager service: <true|false>
+switch_type = kamailio		# Defines the type of switch behind: <freeswitch>
+
+[kamailio]
+# evapi_addr = 127.0.0.1:8448 	# Address of the kamailio evapi server
+# reconnects = 3				# Number of attempts on connect failure.
+
+[mailer]
+# server = localhost								# The server to use when sending emails out
+# auth_user = cgrates								# Authenticate to email server using this user
+# auth_passwd = CGRateS.org							# Authenticate to email server with this password
+# from_address = cgr-mailer@localhost.localdomain	# From address used when sending emails out

data/kamailio/etc/kamailio/kamailio.cfg

@@ -20,7 +20,6 @@ log_facility=LOG_LOCAL0
 fork=yes
 children=4
 listen=172.16.254.79:5060
-#alias="sip.mydomain.com"
 tcp_connection_lifetime=3605
 
 ####### Modules Section ########
@@ -50,6 +49,7 @@ loadmodule "nathelper.so"
 loadmodule "rtpproxy.so"
 loadmodule "htable.so"
 loadmodule "auth.so"
+loadmodule "evapi.so"
 
 
 # ----------------- setting module-specific parameters ---------------
@@ -118,11 +118,24 @@ event_route[htable:mod-init] {
 	$sht(users=>1005) = "check123";
 }
 
+event_route[evapi:connection-new] {
+    xlog("new connection from $evapi(srcaddr):$evapi(srcport)\n");
+}
+
+event_route[evapi:connection-closed] {
+    xlog("connection closed by $evapi(srcaddr):$evapi(srcport)\n");
+}
+
+event_route[evapi:message-received] {
+    xlog("received [$evapi(msg)] from $evapi(srcaddr):$evapi(srcport)\n");
+}
+#evapi_relay("{ \"event\": \"test\",\n \"data\": { \"fU\": \"$fU\" }\n}");
+#evapi_async_relay("{ \"event\": \"suspend\",\n \"data\":"
+#       " { \"index\": \"$T(id_index)\", \"label\": \"$T(id_label)\" }\n}");
+
 # Main SIP request routing logic
 request_route {
 
-	xlog("Request entering server: $mb");
-
 	# per request initial checks
 	route(REQINIT);
 
@@ -182,6 +195,10 @@ request_route {
 
 	# user location service
 	route(LOCATION);
+	#evapi_relay("{ \"event\":\"CGR_AUTHORIZE\",{ \"fU\": \"$fU\" }\n}");
+	evapi_async_relay("{\"event\":\"CGR_AUTHORIZE\",\"tr_index\":\"$T(id_index)\",\"tr_label\":\"$T(id_label)\",\"cgr_account\":\"$fU\",\"cgr_destination\":\"$rU\"}");
+	route(RELAY);
+	exit;
 }
 
 # Wrapper for relaying requests
@@ -294,17 +311,10 @@ route[LOCATION] {
 	if (is_method("INVITE")) {
 		setflag(FLT_ACCMISSED);
 	}
-	route(RELAY);
-	exit;
 }
 
 # user uthentication
 route[AUTH] {
-	xlog("Auth user pwd: $sht(users=>$au) for user: $au");
-	#if (!pv_auth_check("$fd", "$avp(password)", "0", "1")) {
-	#	proxy_challenge("$fd", "1");
-	#	exit;
-	#};
 	if (is_method("REGISTER")) {
 		if ( strempty($au) || !pv_www_authenticate("$td", "$sht(users=>$au)", "0") ) {
 			www_challenge("$td", "0");

data/scripts/pkg/debian/cgrates.init

@@ -64,9 +64,7 @@ do_start()
 	#   0 if daemon has been started
 	#   1 if daemon was already running
 	#   2 if daemon could not be started
-    echo "\n"
-    echo "### Started at:" `date`>>$STACKTRACE
-    echo "\n"
+    echo "\n### Started at:" `date`>>$STACKTRACE
 	start-stop-daemon --start --quiet --pidfile $PIDFILE --exec $DAEMON --test\
                 || return 1
  	start-stop-daemon --start --quiet --chuid $USER:$GROUP --make-pidfile --pidfile $PIDFILE --background\

sessionmanager/fssessionmanager.go

@@ -60,7 +60,9 @@ func (sm *FSSessionManager) Connect() (err error) {
 	} else if !fsock.FS.Connected() {
 		return errors.New("Cannot connect to FreeSWITCH")
 	}
-	fsock.FS.ReadEvents()
+	if err := fsock.FS.ReadEvents(); err != nil {
+		return err
+	}
 	return errors.New("<SessionManager> - Stopped reading events")
 }
 

sessionmanager/kamailiosm.go

@@ -0,0 +1,81 @@
+/*
+Real-time Charging System for Telecom & ISP environments
+Copyright (C) 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 sessionmanager
+
+import (
+	"errors"
+	"fmt"
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
+	"github.com/cgrates/kamevapi"
+	"log/syslog"
+	"regexp"
+	"time"
+)
+
+func NewKamailioSessionManager(cfg *config.CGRConfig, rater, cdrsrv engine.Connector) (*KamailioSessionManager, error) {
+	ksm := &KamailioSessionManager{cgrCfg: cfg, rater: rater, cdrsrv: cdrsrv}
+	return ksm, nil
+}
+
+type KamailioSessionManager struct {
+	cgrCfg        *config.CGRConfig
+	rater         engine.Connector
+	cdrsrv        engine.Connector
+	eventHandlers map[*regexp.Regexp][]func(string)
+	kea           *kamevapi.KamEvapi
+}
+
+func (self *KamailioSessionManager) onCgrAuth(rcvData string) {
+	engine.Logger.Info(fmt.Sprintf("onCgrAuth handler, received: %s\n", rcvData))
+}
+
+func (self *KamailioSessionManager) Connect() error {
+	var err error
+	eventHandlers := map[*regexp.Regexp][]func(string){
+		regexp.MustCompile(".*"): []func(string){self.onCgrAuth},
+	}
+	if self.kea, err = kamevapi.NewKamEvapi(self.cgrCfg.KamailioEvApiAddr, self.cgrCfg.KamailioReconnects, eventHandlers, engine.Logger.(*syslog.Writer)); err != nil {
+		return err
+	}
+	if err := self.kea.ReadEvents(); err != nil {
+		return err
+	}
+	return errors.New("<SM-Kamailio> Stopped reading events")
+}
+
+func (self *KamailioSessionManager) DisconnectSession(uuid, notify, destnr string) {
+	return
+}
+func (self *KamailioSessionManager) RemoveSession(uuid string) {
+	return
+}
+func (self *KamailioSessionManager) MaxDebit(cd *engine.CallDescriptor, cc *engine.CallCost) error {
+	return nil
+}
+func (self *KamailioSessionManager) GetDebitPeriod() time.Duration {
+	var nilDuration time.Duration
+	return nilDuration
+}
+func (self *KamailioSessionManager) GetDbLogger() engine.LogStorage {
+	return nil
+}
+func (self *KamailioSessionManager) Shutdown() error {
+	return nil
+}