X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fregistry.go;h=72a210056efaf6f97d04f9cc779261ffad13ba12;hb=ff383bb22dcf671c68c543294556fe303feb27ff;hp=93a287af8738d6608392b0abdd7bef8c9c102764;hpb=da34eecb23220659f8d519973c03d29444797998;p=ric-plt%2Fsubmgr.git

diff --git a/pkg/control/registry.go b/pkg/control/registry.go
index 93a287a..72a2100 100644
--- a/pkg/control/registry.go
+++ b/pkg/control/registry.go
@@ -20,6 +20,7 @@
 package control
 
 import (
+	"encoding/json"
 	"fmt"
 	"sync"
 	"time"
@@ -37,22 +38,55 @@ type RESTSubscription struct {
 	xAppRmrEndPoint  string
 	Meid             string
 	InstanceIds      []uint32
+	xAppIdToE2Id     map[int64]int64
 	SubReqOngoing    bool
 	SubDelReqOngoing bool
+	lastReqMd5sum    string
 }
 
-func (r *RESTSubscription) AddInstanceId(instanceId uint32) {
+func (r *RESTSubscription) AddE2InstanceId(instanceId uint32) {
+
+	for _, v := range r.InstanceIds {
+		if v == instanceId {
+			return
+		}
+
+	}
+
 	r.InstanceIds = append(r.InstanceIds, instanceId)
 }
 
-func (r *RESTSubscription) SetProcessed() {
-	r.SubReqOngoing = false
+func (r *RESTSubscription) AddMd5Sum(md5sum string) {
+	if md5sum != "" {
+		r.lastReqMd5sum = md5sum
+	} else {
+		xapp.Logger.Error("EMPTY md5sum attempted to be add to subscrition")
+	}
 }
 
-func (r *RESTSubscription) DeleteInstanceId(instanceId uint32) {
+func (r *RESTSubscription) DeleteE2InstanceId(instanceId uint32) {
 	r.InstanceIds = r.InstanceIds[1:]
 }
 
+func (r *RESTSubscription) AddXappIdToE2Id(xAppEventInstanceID int64, e2EventInstanceID int64) {
+	r.xAppIdToE2Id[xAppEventInstanceID] = e2EventInstanceID
+}
+
+func (r *RESTSubscription) GetE2IdFromXappIdToE2Id(xAppEventInstanceID int64) int64 {
+	return r.xAppIdToE2Id[xAppEventInstanceID]
+}
+
+func (r *RESTSubscription) DeleteXappIdToE2Id(xAppEventInstanceID int64) {
+	delete(r.xAppIdToE2Id, xAppEventInstanceID)
+}
+
+func (r *RESTSubscription) SetProcessed(err error) {
+	r.SubReqOngoing = false
+	if err != nil {
+		r.lastReqMd5sum = ""
+	}
+}
+
 type Registry struct {
 	mutex             sync.Mutex
 	register          map[uint32]*Subscription
@@ -71,6 +105,16 @@ func (r *Registry) Initialize() {
 	}
 }
 
+func (r *Registry) GetAllRestSubscriptions() []byte {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	restSubscriptionsJson, err := json.Marshal(r.restSubscriptions)
+	if err != nil {
+		xapp.Logger.Error("GetAllRestSubscriptions(): %v", err)
+	}
+	return restSubscriptionsJson
+}
+
 func (r *Registry) CreateRESTSubscription(restSubId *string, xAppRmrEndPoint *string, maid *string) (*RESTSubscription, error) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
@@ -80,7 +124,8 @@ func (r *Registry) CreateRESTSubscription(restSubId *string, xAppRmrEndPoint *st
 	newRestSubscription.SubReqOngoing = true
 	newRestSubscription.SubDelReqOngoing = false
 	r.restSubscriptions[*restSubId] = &newRestSubscription
-	xapp.Logger.Info("Registry: Created REST subscription successfully. restSubId=%v, subscriptionCount=%v, e2apCubscriptionCount=%v", *restSubId, len(r.restSubscriptions), len(r.register))
+	newRestSubscription.xAppIdToE2Id = make(map[int64]int64)
+	xapp.Logger.Info("Registry: Created REST subscription successfully. restSubId=%v, subscriptionCount=%v, e2apSubscriptionCount=%v", *restSubId, len(r.restSubscriptions), len(r.register))
 	return &newRestSubscription, nil
 }
 
@@ -91,13 +136,15 @@ func (r *Registry) DeleteRESTSubscription(restSubId *string) {
 	xapp.Logger.Info("Registry: Deleted REST subscription successfully. restSubId=%v, subscriptionCount=%v", *restSubId, len(r.restSubscriptions))
 }
 
-func (r *Registry) GetRESTSubscription(restSubId string) (*RESTSubscription, error) {
+func (r *Registry) GetRESTSubscription(restSubId string, IsDelReqOngoing bool) (*RESTSubscription, error) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 	if restSubscription, ok := r.restSubscriptions[restSubId]; ok {
 		// Subscription deletion is not allowed if prosessing subscription request in not ready
 		if restSubscription.SubDelReqOngoing == false && restSubscription.SubReqOngoing == false {
-			restSubscription.SubDelReqOngoing = true
+			if IsDelReqOngoing == true {
+				restSubscription.SubDelReqOngoing = true
+			}
 			r.restSubscriptions[restSubId] = restSubscription
 			return restSubscription, nil
 		} else {
@@ -140,7 +187,7 @@ func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubs
 			NoRespToXapp:     false,
 			DoNotWaitSubResp: false,
 		}
-		subs.ReqId.Id = 123
+		subs.ReqId.Id = subReqMsg.RequestId.Id
 		subs.ReqId.InstanceId = subId
 		if resetTestFlag == true {
 			subs.DoNotWaitSubResp = true
@@ -200,7 +247,8 @@ func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.
 	//
 	actionType, err := r.CheckActionTypes(subReqMsg)
 	if err != nil {
-		xapp.Logger.Debug("CREATE %s", err)
+		xapp.Logger.Info("CREATE %s", err)
+		err = fmt.Errorf("E2 content validation failed")
 		return nil, err
 	}
 
@@ -220,13 +268,16 @@ func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.
 	subs, endPointFound := r.findExistingSubs(trans, subReqMsg)
 	if subs == nil {
 		if subs, err = r.allocateSubs(trans, subReqMsg, resetTestFlag); err != nil {
+			xapp.Logger.Error("%s", err.Error())
+			err = fmt.Errorf("subscription not allocated")
 			return nil, err
 		}
 		newAlloc = true
 	} else if endPointFound == true {
 		// Requesting endpoint is already present in existing subscription. This can happen if xApp is restarted.
 		subs.RetryFromXapp = true
-		xapp.Logger.Debug("CREATE: subscription already exists. %s", subs.String())
+		xapp.Logger.Debug("CREATE subReqMsg.InstanceId=%v. Same subscription %s already exists.", subReqMsg.InstanceId, subs.String())
+		c.UpdateCounter(cDuplicateE2SubReq)
 		return subs, nil
 	}
 
@@ -237,7 +288,7 @@ func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.
 	defer subs.mutex.Unlock()
 
 	epamount := subs.EpList.Size()
-	xapp.Logger.Info("AssignToSubscription subs.EpList.Size() = %v", subs.EpList.Size())
+	xapp.Logger.Info("AssignToSubscription subs.EpList.Size()=%v", subs.EpList.Size())
 
 	r.mutex.Unlock()
 	//
@@ -272,6 +323,8 @@ func (r *Registry) RouteCreate(subs *Subscription, c *Control) error {
 	err := r.rtmgrClient.SubscriptionRequestCreate(subRouteAction)
 	if err != nil {
 		c.UpdateCounter(cRouteCreateFail)
+		xapp.Logger.Error("%s", err.Error())
+		err = fmt.Errorf("RTMGR route create failure")
 	}
 	return err
 }
@@ -281,6 +334,8 @@ func (r *Registry) RouteCreateUpdate(subs *Subscription, c *Control) error {
 	err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 	if err != nil {
 		c.UpdateCounter(cRouteCreateUpdateFail)
+		xapp.Logger.Error("%s", err.Error())
+		err = fmt.Errorf("RTMGR route update failure")
 		return err
 	}
 	c.UpdateCounter(cMergedSubscriptions)
@@ -336,6 +391,7 @@ func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *Transaction
 
 	go func() {
 		if waitRouteClean > 0 {
+			xapp.Logger.Debug("Pending %v in order to wait route cleanup", waitRouteClean)
 			time.Sleep(waitRouteClean)
 		}