SMG - SessionRelocate through the use of InitialOriginID

sessionmanager/smg_event.go

@@ -238,6 +238,18 @@ func (self SMGenericEvent) GetExtraFields() map[string]string {
 	return extraFields
 }
 
+func (self SMGenericEvent) GetFieldAsString(fieldName string) (string, error) {
+	valIf, hasVal := self[fieldName]
+	if !hasVal {
+		return "", utils.ErrNotFound
+	}
+	result, converted := utils.ConvertIfaceToString(valIf)
+	if !converted {
+		return "", utils.ErrNotConvertible
+	}
+	return result, nil
+}
+
 func (self SMGenericEvent) MissingParameter(timezone string) bool {
 	switch self.GetName() {
 	case utils.CGR_AUTHORIZATION:

sessionmanager/smg_event_test.go

@@ -191,3 +191,19 @@ func TestSMGenericEventAsLcrRequest(t *testing.T) {
 		t.Errorf("Expecting: %+v, received: %+v", eLcrReq, lcrReq)
 	}
 }
+
+func TestSMGenericEventGetFieldAsString(t *testing.T) {
+	smGev := SMGenericEvent{}
+	smGev[utils.EVENT_NAME] = "TEST_EVENT"
+	smGev[utils.TOR] = utils.VOICE
+	smGev[utils.ACCID] = "12345"
+	smGev[utils.DIRECTION] = utils.OUT
+	smGev[utils.ACCOUNT] = "account1"
+	smGev[utils.SUBJECT] = "subject1"
+	eFldVal := utils.VOICE
+	if strVal, err := smGev.GetFieldAsString(utils.TOR); err != nil {
+		t.Error(err)
+	} else if strVal != eFldVal {
+		t.Errorf("Expecting: %s, received: %s", eFldVal, strVal)
+	}
+}

sessionmanager/smgeneric.go

@@ -139,6 +139,27 @@ func (self *SMGeneric) sessionEnd(sessionId string, usage time.Duration) error {
 	return err
 }
 
+// Used when an update will relocate an initial session (eg multiple data streams)
+func (self *SMGeneric) sessionRelocate(sessionID, initialID string) error {
+	_, err := self.guard.Guard(func() (interface{}, error) { // Lock it on initialID level
+		if utils.IsSliceMember([]string{sessionID, initialID}, "") {
+			return nil, utils.ErrMandatoryIeMissing
+		}
+		ss := self.getSession(initialID)
+		if len(ss) == 0 { // No need of relocation
+			return nil, utils.ErrNotFound
+		}
+		for i, s := range ss {
+			self.indexSession(sessionID, s)
+			if i == 0 {
+				self.unindexSession(initialID)
+			}
+		}
+		return nil, nil
+	}, time.Duration(2)*time.Second, initialID)
+	return err
+}
+
 // Methods to apply on sessions, mostly exported through RPC/Bi-RPC
 //Calculates maximum usage allowed for gevent
 func (self *SMGeneric) GetMaxUsage(gev SMGenericEvent, clnt *rpc2.Client) (time.Duration, error) {
@@ -170,6 +191,15 @@ func (self *SMGeneric) GetLcrSuppliers(gev SMGenericEvent, clnt *rpc2.Client) ([
 
 // Execute debits for usage/maxUsage
 func (self *SMGeneric) SessionUpdate(gev SMGenericEvent, clnt *rpc2.Client) (time.Duration, error) {
+	if initialID, err := gev.GetFieldAsString(utils.InitialOriginID); err == nil {
+		err := self.sessionRelocate(gev.GetUUID(), initialID)
+		if err == utils.ErrNotFound { // Session was already relocated, create a new  session with this update
+			err = self.sessionStart(gev, getClientConnId(clnt))
+		}
+		if err != nil {
+			return nilDuration, err
+		}
+	}
 	evLastUsed, err := gev.GetLastUsed(utils.META_DEFAULT)
 	if err != nil && err != utils.ErrNotFound {
 		return nilDuration, err

utils/consts.go

@@ -31,6 +31,7 @@ var (
 	ErrAccountDisabled         = errors.New("ACCOUNT_DISABLED")
 	ErrUserNotFound            = errors.New("USER_NOT_FOUND")
 	ErrInsufficientCredit      = errors.New("INSUFFICENT_CREDIT")
+	ErrNotConvertible          = errors.New("NOT_CONVERTIBLE")
 )
 
 const (
@@ -113,6 +114,7 @@ const (
 	TOR                          = "ToR"
 	ORDERID                      = "OrderID"
 	ACCID                        = "OriginID"
+	InitialOriginID              = "InitialOriginID"
 	CDRSOURCE                    = "Source"
 	CDRHOST                      = "OriginHost"
 	REQTYPE                      = "RequestType"