X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=eff14eed912266a3a07e36c9ba9c55161699055a;hb=4abf18056b1674fb284c4d7d753c35a3ddab37e4;hp=1d64f3c64bbf17c77fd2be1235800275b732bbd2;hpb=83ada00338d2c9fa47d48c406b4a46b9d7888aff;p=ric-plt%2Fsubmgr.git

diff --git a/pkg/control/control.go b/pkg/control/control.go
index 1d64f3c..eff14ee 100755
--- a/pkg/control/control.go
+++ b/pkg/control/control.go
@@ -21,34 +21,62 @@ package control
 
 import (
 	"fmt"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
 	"gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap"
 	rtmgrclient "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client"
+	"gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models"
 	"gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
 	httptransport "github.com/go-openapi/runtime/client"
 	"github.com/go-openapi/strfmt"
+	"github.com/gorilla/mux"
 	"github.com/spf13/viper"
-	"sync"
-	"time"
 )
 
 //-----------------------------------------------------------------------------
 //
 //-----------------------------------------------------------------------------
 
-var e2tSubReqTimeout time.Duration = 5 * time.Second
-var e2tSubDelReqTime time.Duration = 5 * time.Second
-var e2tMaxSubReqTryCount uint64 = 2    // Initial try + retry
-var e2tMaxSubDelReqTryCount uint64 = 2 // Initial try + retry
+func idstring(err error, entries ...fmt.Stringer) string {
+	var retval string = ""
+	var filler string = ""
+	for _, entry := range entries {
+		retval += filler + entry.String()
+		filler = " "
+	}
+	if err != nil {
+		retval += filler + "err(" + err.Error() + ")"
+		filler = " "
+
+	}
+	return retval
+}
 
-var e2tRecvMsgTimeout time.Duration = 5 * time.Second
+//-----------------------------------------------------------------------------
+//
+//-----------------------------------------------------------------------------
+
+var e2tSubReqTimeout time.Duration
+var e2tSubDelReqTime time.Duration
+var e2tRecvMsgTimeout time.Duration
+var e2tMaxSubReqTryCount uint64    // Initial try + retry
+var e2tMaxSubDelReqTryCount uint64 // Initial try + retry
+var readSubsFromDb string
 
 type Control struct {
-	e2ap         *E2ap
-	registry     *Registry
-	tracker      *Tracker
-	timerMap     *TimerMap
-	rmrSendMutex sync.Mutex
-	msgCounter   uint64
+	*xapp.RMRClient
+	e2ap     *E2ap
+	registry *Registry
+	tracker  *Tracker
+	db       Sdlnterface
+	//subscriber *xapp.Subscriber
+	CntRecvMsg    uint64
+	ResetTestFlag bool
+	Counters      map[string]xapp.Counter
 }
 
 type RMRMeid struct {
@@ -57,6 +85,9 @@ type RMRMeid struct {
 	RanName string
 }
 
+type SubmgrRestartTestEvent struct{}
+type SubmgrRestartUpEvent struct{}
+
 func init() {
 	xapp.Logger.Info("SUBMGR")
 	viper.AutomaticEnv()
@@ -66,6 +97,7 @@ func init() {
 
 func NewControl() *Control {
 
+	ReadConfigParameters()
 	transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"})
 	rtmgrClient := RtmgrClient{rtClient: rtmgrclient.New(transport, strfmt.Default)}
 
@@ -76,118 +108,243 @@ func NewControl() *Control {
 	tracker := new(Tracker)
 	tracker.Init()
 
-	timerMap := new(TimerMap)
-	timerMap.Init()
+	//subscriber := xapp.NewSubscriber(viper.GetString("subscription.host"), viper.GetInt("subscription.timeout"))
+
+	c := &Control{e2ap: new(E2ap),
+		registry: registry,
+		tracker:  tracker,
+		db:       CreateSdl(),
+		//subscriber: subscriber,
+		Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"),
+	}
+
+	// Register REST handler for testing support
+	xapp.Resource.InjectRoute("/ric/v1/test/{testId}", c.TestRestHandler, "POST")
 
-	return &Control{e2ap: new(E2ap),
-		registry:   registry,
-		tracker:    tracker,
-		timerMap:   timerMap,
-		msgCounter: 0,
+	go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandler)
+	//go c.subscriber.Listen(c.SubscriptionHandler, c.QueryHandler)
+
+	if readSubsFromDb == "false" {
+		return c
+	}
+
+	// Read subscriptions from db
+	xapp.Logger.Info("Reading subscriptions from db")
+	subIds, register, err := c.ReadAllSubscriptionsFromSdl()
+	if err != nil {
+		xapp.Logger.Error("%v", err)
+	} else {
+		c.registry.subIds = subIds
+		c.registry.register = register
+		c.HandleUncompletedSubscriptions(register)
+	}
+	return c
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func ReadConfigParameters() {
+
+	// viper.GetDuration returns nanoseconds
+	e2tSubReqTimeout = viper.GetDuration("controls.e2tSubReqTimeout_ms") * 1000000
+	if e2tSubReqTimeout == 0 {
+		e2tSubReqTimeout = 2000 * 1000000
+	}
+	xapp.Logger.Info("e2tSubReqTimeout %v", e2tSubReqTimeout)
+	e2tSubDelReqTime = viper.GetDuration("controls.e2tSubDelReqTime_ms") * 1000000
+	if e2tSubDelReqTime == 0 {
+		e2tSubDelReqTime = 2000 * 1000000
+	}
+	xapp.Logger.Info("e2tSubDelReqTime %v", e2tSubDelReqTime)
+	e2tRecvMsgTimeout = viper.GetDuration("controls.e2tRecvMsgTimeout_ms") * 1000000
+	if e2tRecvMsgTimeout == 0 {
+		e2tRecvMsgTimeout = 2000 * 1000000
+	}
+	xapp.Logger.Info("e2tRecvMsgTimeout %v", e2tRecvMsgTimeout)
+	e2tMaxSubReqTryCount = viper.GetUint64("controls.e2tMaxSubReqTryCount")
+	if e2tMaxSubReqTryCount == 0 {
+		e2tMaxSubReqTryCount = 1
+	}
+	xapp.Logger.Info("e2tMaxSubReqTryCount %v", e2tMaxSubReqTryCount)
+	e2tMaxSubDelReqTryCount = viper.GetUint64("controls.e2tMaxSubDelReqTryCount")
+	if e2tMaxSubDelReqTryCount == 0 {
+		e2tMaxSubDelReqTryCount = 1
+	}
+	xapp.Logger.Info("e2tMaxSubDelReqTryCount %v", e2tMaxSubDelReqTryCount)
+
+	readSubsFromDb = viper.GetString("controls.readSubsFromDb")
+	if readSubsFromDb == "" {
+		readSubsFromDb = "true"
+	}
+	xapp.Logger.Info("readSubsFromDb %v", readSubsFromDb)
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) HandleUncompletedSubscriptions(register map[uint32]*Subscription) {
+
+	xapp.Logger.Debug("HandleUncompletedSubscriptions. len(register) = %v", len(register))
+	for subId, subs := range register {
+		if subs.SubRespRcvd == false {
+			subs.NoRespToXapp = true
+			xapp.Logger.Debug("SendSubscriptionDeleteReq. subId = %v", subId)
+			c.SendSubscriptionDeleteReq(subs)
+		}
+	}
+}
+
+func (c *Control) ReadyCB(data interface{}) {
+	if c.RMRClient == nil {
+		c.RMRClient = xapp.Rmr
 	}
 }
 
 func (c *Control) Run() {
+	xapp.SetReadyCB(c.ReadyCB, nil)
 	xapp.Run(c)
 }
 
-func (c *Control) rmrSendRaw(desc string, params *RMRParams) (err error) {
-
-	xapp.Logger.Info("%s: %s", desc, params.String())
-	status := false
-	i := 1
-	for ; i <= 10 && status == false; i++ {
-		c.rmrSendMutex.Lock()
-		status = xapp.Rmr.Send(params.RMRParams, false)
-		c.rmrSendMutex.Unlock()
-		if status == false {
-			xapp.Logger.Info("rmr.Send() failed. Retry count %d, %s", i, params.String())
-			time.Sleep(500 * time.Millisecond)
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) SubscriptionHandler(stype models.SubscriptionType, params interface{}) (*models.SubscriptionResponse, error) {
+	/*
+	   switch p := params.(type) {
+	   case *models.ReportParams:
+	       trans := c.tracker.NewXappTransaction(NewRmrEndpoint(p.ClientEndpoint),"" , 0, &xapp.RMRMeid{RanName: p.Meid})
+	       if trans == nil {
+	             xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
+	             return
+	       }
+	       defer trans.Release()
+	   case *models.ControlParams:
+	   case *models.PolicyParams:
+	   }
+	*/
+	return &models.SubscriptionResponse{}, fmt.Errorf("Subscription rest interface not implemented")
+}
+
+func (c *Control) SubscriptionDeleteHandler(s string) error {
+	return nil
+}
+
+func (c *Control) QueryHandler() (models.SubscriptionList, error) {
+	return c.registry.QueryHandler()
+}
+
+func (c *Control) TestRestHandler(w http.ResponseWriter, r *http.Request) {
+
+	xapp.Logger.Info("TestRestHandler() called")
+
+	pathParams := mux.Vars(r)
+	s := pathParams["testId"]
+
+	// This can be used to delete single subscription from db
+	if contains := strings.Contains(s, "deletesubid="); contains == true {
+		var splits = strings.Split(s, "=")
+		if subId, err := strconv.ParseInt(splits[1], 10, 64); err == nil {
+			xapp.Logger.Info("RemoveSubscriptionFromSdl() called. subId = %v", subId)
+			c.RemoveSubscriptionFromSdl(uint32(subId))
+			return
 		}
 	}
-	if status == false {
-		err = fmt.Errorf("rmr.Send() failed. Retry count %d, %s", i, params.String())
-		xapp.Logger.Error("%s: %s", desc, err.Error())
-		xapp.Rmr.Free(params.Mbuf)
+
+	// This can be used to remove all subscriptions db from
+	if s == "emptydb" {
+		xapp.Logger.Info("RemoveAllSubscriptionsFromSdl() called")
+		c.RemoveAllSubscriptionsFromSdl()
+		return
 	}
-	return
+
+	// This is meant to cause submgr's restart in testing
+	if s == "restart" {
+		xapp.Logger.Info("os.Exit(1) called")
+		os.Exit(1)
+	}
+
+	xapp.Logger.Info("Unsupported rest command received %s", s)
 }
 
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+
 func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *TransactionSubs) (err error) {
-	params := &RMRParams{&xapp.RMRParams{}}
+	params := &xapp.RMRParams{}
 	params.Mtype = trans.GetMtype()
-	params.SubId = int(subs.GetReqId().Seq)
+	params.SubId = int(subs.GetReqId().InstanceId)
 	params.Xid = ""
 	params.Meid = subs.GetMeid()
 	params.Src = ""
 	params.PayloadLen = len(trans.Payload.Buf)
 	params.Payload = trans.Payload.Buf
 	params.Mbuf = nil
-
-	return c.rmrSendRaw("MSG to E2T:"+desc+":"+trans.String(), params)
+	xapp.Logger.Info("MSG to E2T: %s %s %s", desc, trans.String(), params.String())
+	return c.SendWithRetry(params, false, 5)
 }
 
 func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) {
 
-	params := &RMRParams{&xapp.RMRParams{}}
+	params := &xapp.RMRParams{}
 	params.Mtype = trans.GetMtype()
-	params.SubId = int(subs.GetReqId().Seq)
+	params.SubId = int(subs.GetReqId().InstanceId)
 	params.Xid = trans.GetXid()
 	params.Meid = trans.GetMeid()
 	params.Src = ""
 	params.PayloadLen = len(trans.Payload.Buf)
 	params.Payload = trans.Payload.Buf
 	params.Mbuf = nil
-
-	return c.rmrSendRaw("MSG to XAPP:"+desc+":"+trans.String(), params)
+	xapp.Logger.Info("MSG to XAPP: %s %s %s", desc, trans.String(), params.String())
+	return c.SendWithRetry(params, false, 5)
 }
 
-func (c *Control) Consume(params *xapp.RMRParams) (err error) {
-	xapp.Rmr.Free(params.Mbuf)
-	params.Mbuf = nil
-	msg := &RMRParams{params}
-	c.msgCounter++
+func (c *Control) Consume(msg *xapp.RMRParams) (err error) {
+	if c.RMRClient == nil {
+		err = fmt.Errorf("Rmr object nil can handle %s", msg.String())
+		xapp.Logger.Error("%s", err.Error())
+		return
+	}
+	c.CntRecvMsg++
+
+	defer c.RMRClient.Free(msg.Mbuf)
+
+	// xapp-frame might use direct access to c buffer and
+	// when msg.Mbuf is freed, someone might take it into use
+	// and payload data might be invalid inside message handle function
+	//
+	// subscriptions won't load system a lot so there is no
+	// real performance hit by cloning buffer into new go byte slice
+	cPay := append(msg.Payload[:0:0], msg.Payload...)
+	msg.Payload = cPay
+	msg.PayloadLen = len(cPay)
 
 	switch msg.Mtype {
-	case xapp.RICMessageTypes["RIC_SUB_REQ"]:
+	case xapp.RIC_SUB_REQ:
 		go c.handleXAPPSubscriptionRequest(msg)
-	case xapp.RICMessageTypes["RIC_SUB_RESP"]:
+	case xapp.RIC_SUB_RESP:
 		go c.handleE2TSubscriptionResponse(msg)
-	case xapp.RICMessageTypes["RIC_SUB_FAILURE"]:
+	case xapp.RIC_SUB_FAILURE:
 		go c.handleE2TSubscriptionFailure(msg)
-	case xapp.RICMessageTypes["RIC_SUB_DEL_REQ"]:
+	case xapp.RIC_SUB_DEL_REQ:
 		go c.handleXAPPSubscriptionDeleteRequest(msg)
-	case xapp.RICMessageTypes["RIC_SUB_DEL_RESP"]:
+	case xapp.RIC_SUB_DEL_RESP:
 		go c.handleE2TSubscriptionDeleteResponse(msg)
-	case xapp.RICMessageTypes["RIC_SUB_DEL_FAILURE"]:
+	case xapp.RIC_SUB_DEL_FAILURE:
 		go c.handleE2TSubscriptionDeleteFailure(msg)
 	default:
 		xapp.Logger.Info("Unknown Message Type '%d', discarding", msg.Mtype)
 	}
-
-	return nil
-}
-
-func idstring(err error, entries ...fmt.Stringer) string {
-	var retval string = ""
-	var filler string = ""
-	for _, entry := range entries {
-		retval += filler + entry.String()
-		filler = " "
-	}
-	if err != nil {
-		retval += filler + "err(" + err.Error() + ")"
-		filler = " "
-
-	}
-	return retval
+	return
 }
 
 //-------------------------------------------------------------------
 // handle from XAPP Subscription Request
 //------------------------------------------------------------------
-func (c *Control) handleXAPPSubscriptionRequest(params *RMRParams) {
+func (c *Control) handleXAPPSubscriptionRequest(params *xapp.RMRParams) {
 	xapp.Logger.Info("MSG from XAPP: %s", params.String())
+	c.UpdateCounter(cSubReqFromXapp)
 
 	subReqMsg, err := c.e2ap.UnpackSubscriptionRequest(params.Payload)
 	if err != nil {
@@ -195,7 +352,7 @@ func (c *Control) handleXAPPSubscriptionRequest(params *RMRParams) {
 		return
 	}
 
-	trans := c.tracker.NewXappTransaction(NewRmrEndpoint(params.Src), params.Xid, &RequestId{subReqMsg.RequestId}, params.Meid)
+	trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subReqMsg.RequestId.InstanceId, params.Meid)
 	if trans == nil {
 		xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 		return
@@ -208,7 +365,8 @@ func (c *Control) handleXAPPSubscriptionRequest(params *RMRParams) {
 		return
 	}
 
-	subs, err := c.registry.AssignToSubscription(trans, subReqMsg)
+	//TODO handle subscription toward e2term inside AssignToSubscription / hide handleSubscriptionCreate in it?
+	subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c)
 	if err != nil {
 		xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 		return
@@ -219,19 +377,21 @@ func (c *Control) handleXAPPSubscriptionRequest(params *RMRParams) {
 	//
 	go c.handleSubscriptionCreate(subs, trans)
 	event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
-
 	err = nil
 	if event != nil {
 		switch themsg := event.(type) {
 		case *e2ap.E2APSubscriptionResponse:
 			trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(themsg)
 			if err == nil {
+				trans.Release()
+				c.UpdateCounter(cSubRespToXapp)
 				c.rmrSendToXapp("", subs, trans)
 				return
 			}
 		case *e2ap.E2APSubscriptionFailure:
 			trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(themsg)
 			if err == nil {
+				c.UpdateCounter(cSubFailToXapp)
 				c.rmrSendToXapp("", subs, trans)
 			}
 		default:
@@ -239,14 +399,15 @@ func (c *Control) handleXAPPSubscriptionRequest(params *RMRParams) {
 		}
 	}
 	xapp.Logger.Info("XAPP-SubReq: failed %s", idstring(err, trans, subs))
-	go c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
+	//c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 }
 
 //-------------------------------------------------------------------
 // handle from XAPP Subscription Delete Request
 //------------------------------------------------------------------
-func (c *Control) handleXAPPSubscriptionDeleteRequest(params *RMRParams) {
+func (c *Control) handleXAPPSubscriptionDeleteRequest(params *xapp.RMRParams) {
 	xapp.Logger.Info("MSG from XAPP: %s", params.String())
+	c.UpdateCounter(cSubDelReqFromXapp)
 
 	subDelReqMsg, err := c.e2ap.UnpackSubscriptionDeleteRequest(params.Payload)
 	if err != nil {
@@ -254,7 +415,7 @@ func (c *Control) handleXAPPSubscriptionDeleteRequest(params *RMRParams) {
 		return
 	}
 
-	trans := c.tracker.NewXappTransaction(NewRmrEndpoint(params.Src), params.Xid, &RequestId{subDelReqMsg.RequestId}, params.Meid)
+	trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subDelReqMsg.RequestId.InstanceId, params.Meid)
 	if trans == nil {
 		xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 		return
@@ -267,7 +428,7 @@ func (c *Control) handleXAPPSubscriptionDeleteRequest(params *RMRParams) {
 		return
 	}
 
-	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelReqMsg.RequestId.Seq})
+	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{trans.GetSubId()})
 	if err != nil {
 		xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(err, trans))
 		return
@@ -281,16 +442,23 @@ func (c *Control) handleXAPPSubscriptionDeleteRequest(params *RMRParams) {
 
 	xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
 
-	// Whatever is received send ok delete response
+	if subs.NoRespToXapp == true {
+		// Do no send delete responses to xapps due to submgr restart is deleting uncompleted subscriptions
+		return
+	}
+
+	// Whatever is received success, fail or timeout, send successful delete response
 	subDelRespMsg := &e2ap.E2APSubscriptionDeleteResponse{}
-	subDelRespMsg.RequestId = subs.SubReqMsg.RequestId
+	subDelRespMsg.RequestId = subs.GetReqId().RequestId
 	subDelRespMsg.FunctionId = subs.SubReqMsg.FunctionId
 	trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(subDelRespMsg)
 	if err == nil {
+		c.UpdateCounter(cSubDelRespToXapp)
 		c.rmrSendToXapp("", subs, trans)
 	}
 
-	go c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
+	//TODO handle subscription toward e2term insiged RemoveFromSubscription / hide handleSubscriptionDelete in it?
+	//c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 }
 
 //-------------------------------------------------------------------
@@ -298,6 +466,7 @@ func (c *Control) handleXAPPSubscriptionDeleteRequest(params *RMRParams) {
 //-------------------------------------------------------------------
 func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *TransactionXapp) {
 
+	var removeSubscriptionFromDb bool = false
 	trans := c.tracker.NewSubsTransaction(subs)
 	subs.WaitTransactionTurn(trans)
 	defer subs.ReleaseTransactionTurn(trans)
@@ -311,10 +480,19 @@ func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *Tran
 		switch event.(type) {
 		case *e2ap.E2APSubscriptionResponse:
 			subRfMsg, valid = subs.SetCachedResponse(event, true)
+			subs.SubRespRcvd = true
 		case *e2ap.E2APSubscriptionFailure:
+			removeSubscriptionFromDb = true
 			subRfMsg, valid = subs.SetCachedResponse(event, false)
+			xapp.Logger.Info("SUBS-SubReq: internal delete  due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
+			c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
+		case *SubmgrRestartTestEvent:
+			// This simulates that no response has been received and after restart subscriptions are restored from db
+			xapp.Logger.Debug("Test restart flag is active. Dropping this transaction to test restart case")
+			return
 		default:
 			xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
+			removeSubscriptionFromDb = true
 			subRfMsg, valid = subs.SetCachedResponse(nil, false)
 			c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 		}
@@ -323,6 +501,12 @@ func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *Tran
 		xapp.Logger.Debug("SUBS-SubReq: Handling (cached response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 	}
 
+	//Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
+	if valid == false {
+		c.registry.RemoveFromSubscription(subs, parentTrans, 5*time.Second, c)
+	}
+
+	c.UpdateSubscriptionInDB(subs, removeSubscriptionFromDb)
 	parentTrans.SendEvent(subRfMsg, 0)
 }
 
@@ -340,6 +524,7 @@ func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *Tran
 	xapp.Logger.Debug("SUBS-SubDelReq: Handling %s", idstring(nil, trans, subs, parentTrans))
 
 	subs.mutex.Lock()
+
 	if subs.valid && subs.EpList.HasEndpoint(parentTrans.GetEndpoint()) && subs.EpList.Size() == 1 {
 		subs.valid = false
 		subs.mutex.Unlock()
@@ -347,11 +532,12 @@ func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *Tran
 	} else {
 		subs.mutex.Unlock()
 	}
-
-	subDelRespMsg := &e2ap.E2APSubscriptionDeleteResponse{}
-	subDelRespMsg.RequestId = subs.SubReqMsg.RequestId
-	subDelRespMsg.FunctionId = subs.SubReqMsg.FunctionId
-	parentTrans.SendEvent(subDelRespMsg, 0)
+	//Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
+	//  If parallel deletes ongoing both might pass earlier sendE2TSubscriptionDeleteRequest(...) if
+	//  RemoveFromSubscription locates in caller side (now in handleXAPPSubscriptionDeleteRequest(...))
+	c.registry.RemoveFromSubscription(subs, parentTrans, 5*time.Second, c)
+	c.registry.UpdateSubscriptionToDb(subs, c)
+	parentTrans.SendEvent(nil, 0)
 }
 
 //-------------------------------------------------------------------
@@ -370,12 +556,25 @@ func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *Transact
 		return event
 	}
 
+	// Write uncompleted subscrition in db. If no response for subscrition it need to be re-processed (deleted) after restart
+	c.WriteSubscriptionToDb(subs)
 	for retries := uint64(0); retries < e2tMaxSubReqTryCount; retries++ {
 		desc := fmt.Sprintf("(retry %d)", retries)
+		if retries == 0 {
+			c.UpdateCounter(cSubReqToE2)
+		} else {
+			c.UpdateCounter(cSubReReqToE2)
+		}
 		c.rmrSendToE2T(desc, subs, trans)
-		event, timedOut = trans.WaitEvent(e2tSubReqTimeout)
-		if timedOut {
-			continue
+		if subs.DoNotWaitSubResp == false {
+			event, timedOut = trans.WaitEvent(e2tSubReqTimeout)
+			if timedOut {
+				c.UpdateCounter(cSubReqTimerExpiry)
+				continue
+			}
+		} else {
+			// Simulating case where subscrition request has been sent but response has not been received before restart
+			event = &SubmgrRestartTestEvent{}
 		}
 		break
 	}
@@ -403,9 +602,15 @@ func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *Tr
 
 	for retries := uint64(0); retries < e2tMaxSubDelReqTryCount; retries++ {
 		desc := fmt.Sprintf("(retry %d)", retries)
+		if retries == 0 {
+			c.UpdateCounter(cSubDelReqToE2)
+		} else {
+			c.UpdateCounter(cSubDelReReqToE2)
+		}
 		c.rmrSendToE2T(desc, subs, trans)
 		event, timedOut = trans.WaitEvent(e2tSubDelReqTime)
 		if timedOut {
+			c.UpdateCounter(cSubDelReqTimerExpiry)
 			continue
 		}
 		break
@@ -415,16 +620,17 @@ func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *Tr
 }
 
 //-------------------------------------------------------------------
-// handle from E2T Subscription Reponse
+// handle from E2T Subscription Response
 //-------------------------------------------------------------------
-func (c *Control) handleE2TSubscriptionResponse(params *RMRParams) {
+func (c *Control) handleE2TSubscriptionResponse(params *xapp.RMRParams) {
 	xapp.Logger.Info("MSG from E2T: %s", params.String())
+	c.UpdateCounter(cSubRespFromE2)
 	subRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload)
 	if err != nil {
 		xapp.Logger.Error("MSG-SubResp %s", idstring(err, params))
 		return
 	}
-	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subRespMsg.RequestId.Seq})
+	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subRespMsg.RequestId.InstanceId})
 	if err != nil {
 		xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params))
 		return
@@ -446,14 +652,15 @@ func (c *Control) handleE2TSubscriptionResponse(params *RMRParams) {
 //-------------------------------------------------------------------
 // handle from E2T Subscription Failure
 //-------------------------------------------------------------------
-func (c *Control) handleE2TSubscriptionFailure(params *RMRParams) {
+func (c *Control) handleE2TSubscriptionFailure(params *xapp.RMRParams) {
 	xapp.Logger.Info("MSG from E2T: %s", params.String())
+	c.UpdateCounter(cSubFailFromE2)
 	subFailMsg, err := c.e2ap.UnpackSubscriptionFailure(params.Payload)
 	if err != nil {
 		xapp.Logger.Error("MSG-SubFail %s", idstring(err, params))
 		return
 	}
-	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subFailMsg.RequestId.Seq})
+	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subFailMsg.RequestId.InstanceId})
 	if err != nil {
 		xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params))
 		return
@@ -475,14 +682,15 @@ func (c *Control) handleE2TSubscriptionFailure(params *RMRParams) {
 //-------------------------------------------------------------------
 // handle from E2T Subscription Delete Response
 //-------------------------------------------------------------------
-func (c *Control) handleE2TSubscriptionDeleteResponse(params *RMRParams) (err error) {
+func (c *Control) handleE2TSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
 	xapp.Logger.Info("MSG from E2T: %s", params.String())
+	c.UpdateCounter(cSubDelRespFromE2)
 	subDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload)
 	if err != nil {
 		xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 		return
 	}
-	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelRespMsg.RequestId.Seq})
+	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelRespMsg.RequestId.InstanceId})
 	if err != nil {
 		xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 		return
@@ -504,14 +712,15 @@ func (c *Control) handleE2TSubscriptionDeleteResponse(params *RMRParams) (err er
 //-------------------------------------------------------------------
 // handle from E2T Subscription Delete Failure
 //-------------------------------------------------------------------
-func (c *Control) handleE2TSubscriptionDeleteFailure(params *RMRParams) {
+func (c *Control) handleE2TSubscriptionDeleteFailure(params *xapp.RMRParams) {
 	xapp.Logger.Info("MSG from E2T: %s", params.String())
+	c.UpdateCounter(cSubDelFailFromE2)
 	subDelFailMsg, err := c.e2ap.UnpackSubscriptionDeleteFailure(params.Payload)
 	if err != nil {
 		xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
 		return
 	}
-	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelFailMsg.RequestId.Seq})
+	subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelFailMsg.RequestId.InstanceId})
 	if err != nil {
 		xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
 		return
@@ -554,3 +763,76 @@ func typeofSubsMessage(v interface{}) string {
 		return "Unknown"
 	}
 }
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) WriteSubscriptionToDb(subs *Subscription) {
+	xapp.Logger.Debug("WriteSubscriptionToDb() subId = %v", subs.ReqId.InstanceId)
+	err := c.WriteSubscriptionToSdl(subs.ReqId.InstanceId, subs)
+	if err != nil {
+		xapp.Logger.Error("%v", err)
+	}
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) UpdateSubscriptionInDB(subs *Subscription, removeSubscriptionFromDb bool) {
+
+	if removeSubscriptionFromDb == true {
+		// Subscription was written in db already when subscription request was sent to BTS, except for merged request
+		c.RemoveSubscriptionFromDb(subs)
+	} else {
+		// Update is needed for successful response and merge case here
+		if subs.RetryFromXapp == false {
+			c.WriteSubscriptionToDb(subs)
+		}
+	}
+	subs.RetryFromXapp = false
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) RemoveSubscriptionFromDb(subs *Subscription) {
+	xapp.Logger.Debug("RemoveSubscriptionFromDb() subId = %v", subs.ReqId.InstanceId)
+	err := c.RemoveSubscriptionFromSdl(subs.ReqId.InstanceId)
+	if err != nil {
+		xapp.Logger.Error("%v", err)
+	}
+}
+
+func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) {
+
+	xapp.Logger.Debug("Sending subscription delete due to restart. subId = %v", subs.ReqId.InstanceId)
+
+	// Send delete for every endpoint in the subscription
+	subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
+	subDelReqMsg.RequestId = subs.GetReqId().RequestId
+	subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
+	mType, payload, err := c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
+	if err != nil {
+		xapp.Logger.Error("SendSubscriptionDeleteReq() %s", idstring(err))
+		return
+	}
+	for _, endPoint := range subs.EpList.Endpoints {
+		params := &xapp.RMRParams{}
+		params.Mtype = mType
+		params.SubId = int(subs.GetReqId().InstanceId)
+		params.Xid = ""
+		params.Meid = subs.Meid
+		params.Src = endPoint.String()
+		params.PayloadLen = len(payload.Buf)
+		params.Payload = payload.Buf
+		params.Mbuf = nil
+
+		if params == nil {
+			xapp.Logger.Error("SendSubscriptionDeleteReq() params == nil")
+			return
+		}
+
+		subs.DeleteFromDb = true
+		c.handleXAPPSubscriptionDeleteRequest(params)
+	}
+}